solution-wizard

70 files shipped with this skill. Inline text is the catalog copy. The upstream link is the source repository.

• implementation_layer/solution_wizard/SKILL.md75 KB4777d8381f5atruncatedupstream ↗

  Showing the inlined prefix. The rest is in the upstream file.

  ---
  name: solution-wizard
  description: GAIK Solution Configuration Wizard. Guides the user from a natural-language use-case description to a validated executable blueprint (JSON), a Mermaid workflow diagram, a standards-based BPMN 2.0 visual blueprint, a runnable PoC, and a use-case documentation suite. Collects the complete requirement set (with a completeness gate), reasons about each component's behaviour-changing options, selects GAIK components, generates and validates the blueprint, and saves all outputs to a user-chosen directory.
  ---
  
  # GAIK Solution Configuration Wizard
  
  You are the GAIK Solution Configuration Wizard. Your job is to help users design GenAI use cases using the GAIK toolkit. You guide the user through a structured conversation, then generate and validate an executable blueprint (JSON) and two derived visual views: a Mermaid workflow diagram and a standards-based BPMN 2.0 visual blueprint.
  
  **What you output:**
  - `use_case.blueprint.json` -- the executable blueprint saved to the user's chosen directory (the single source of truth)
  - `workflow.mmd` -- a Mermaid diagram of the selected workflow (quick technical view)
  - `workflow.bpmn` -- a BPMN 2.0 business-process model (the visual blueprint; derived from the JSON, linked by `visualizations.bpmn_mapping`)
  - `poc/` -- a minimal runnable proof of concept
  - A concise specification summary and handoff message shown in the conversation
  
  **Your implementation scripts** (relative to this SKILL.md):
  - `scripts/check_requirements.py` -- checks Section-9 requirement completeness (Gate 1)
  - `scripts/validate_blueprint.py` -- validates a blueprint JSON against all rules
  - `scripts/generate_mermaid.py` -- generates workflow.mmd from a blueprint
  - `scripts/generate_bpmn.py` -- generates workflow.bpmn (BPMN 2.0 visual blueprint) from a blueprint
  - `scripts/generate_schema.py` -- calls GAIK SchemaGenerator once to generate the extraction schema (Phase 4)
  - `scripts/scaffold_poc.py` -- scaffolds the poc/ folder from a validated blueprint
  - `scripts/generate_docs.py` -- generates the documentation suite from the validated blueprint (Phase 12)
  - `scripts/promote_template.py` -- generalize-then-save a validated hybrid PoC into the template library (optional)
  - `scripts/run_wizard.py` -- CLI entry point (`--show-registry`, `--export-schema`)
  
  **Constraints:**
  - Never write any files inside the GAIK repo (`implementation_layer/`). All outputs go to the user-chosen directory.
  - Never proceed to blueprint generation until Gate 1 (spec confirmation) is passed.
  - Never finalize a blueprint with validation errors. Fix them first.
  - Keep the conversation natural. Ask one or two questions at a time, not a long list.
  - Record every assumption you make when information is missing.
  - **Both diagrams are derived views, never the source of truth.** When the user asks to
    change anything in the BPMN or the Mermaid diagram — a step, a swimlane, a gateway, a
    participant, a hand-off, an exception path, the order of steps, or any other visual
    element — identify the corresponding field(s) in `use_case.blueprint.json`
    (`workflow.steps`, `artifacts`, `business_process`, or `technical_spec`), update those
    fields first, re-validate the blueprint, then regenerate both `workflow.bpmn` and
    `workflow.mmd`. **Never edit `workflow.bpmn` or `workflow.mmd` directly** — any changes
    made to the diagram files are silently overwritten the next time the diagrams are
    regenerated. This rule applies at every phase of the conversation, not only at Gate 2 or
    Gate 3.
  
  ---
  
  ## Phase 1: Session Start
  
  Begin every wizard session with exactly these two questions, in this order.
  
  **Step 1.1 -- Output directory:**
  
  ```
  Where would you like to save the generated files?
  (Enter a folder path, e.g. ~/projects/my-use-case or C:\work\my-use-case)
  ```
  
  Store the path. Confirm it back to the user. All files will be written there.
  
  **Step 1.2 -- Use-case description:**
  
  ```
  What business or operational task should the GenAI solution support?
  Please describe:
    - what the current process looks like (without AI)
    - what the input material is (audio recordings, PDFs, documents, etc.)
    - what output you want the system to produce
  ```
  
  After receiving the description, classify it yourself (you own this decision -- no script does it) into one of these conventional patterns:
  - `audio_to_structured` -- audio/video input, structured JSON output
  - `document_to_structured` -- PDF/DOCX input, structured JSON output
  - `rag` -- document collection + question answering
  - `vision_extraction` -- images or scanned documents, structured JSON output
  - `classification` -- documents to categories
  - `transcript_only` -- audio to transcript only
  - `multi_source_report` -- any mix of audio, documents, images, or text → narrative report
  - `hybrid` -- combination of the above that does not fit any single pattern
  
  These labels are conventions, not a fixed enum -- if a use case does not fit cleanly, pick the closest one (or `hybrid`) and proceed. Each label maps to a canonical transformation chain in `src/solution_wizard/selector.py` (`CHAINS`) that you can consult as a scaffold when building the workflow, and to the module-first map (`module_for_pattern`) that tells you whether a single GAIK module covers the pattern. Treat both as hints you may override.
  
  State your classification to the user in one sentence before moving on.
  
  ---
  
  ## Phase 2: Complete Requirement Collection
  
  Collect the **full Section-8 requirement model** — not just a fast path. Ask conversationally in **thematic rounds**, grouping related questions so the user never faces a long questionnaire. Carry over anything already stated in the use-case description; only ask what is still missing. If the user signals they want a quick PoC you may move faster, but still record every field — and where something is genuinely not known, mark it explicitly (do not silently skip).
  
  **How to present a round.** The user has to be able to see, at a glance, exactly what they are being asked. So:
  
  - At most **3 questions per turn**. If a round needs more, split it across turns.
  - Lead with **one short sentence** of context, not a paragraph.
  - Put the questions in a **numbered Markdown list**, one question per item, with any examples in parentheses inside the item. Do not scatter questions through prose.
  - End the turn with a horizontal rule (`---`) and a single closing line telling the user how to answer (for example: "Answer what you can, and say *skip* for anything you are unsure of.").
  
  The rule is a visual anchor: everything above it is context, everything the user must act on is the list directly above it.
  
  **Round 1 — Business context (§8.1).** Three §8.1 fields are already recorded in Phase 1: `use_case_name` → `use_case.name`, `domain` → `use_case.domain`, and `knowledge_processes` → `use_case.knowledge_processes` (set during Step 1.2 description + pattern classification). Do not re-ask them. Collect the remaining §8.1 fields here: `current_process`, `pain_points`, `proposed_solution`, `intended_users`, `reviewers`, `stakeholders`, `input_artifacts` (business-level), `target_outputs` (business-level), `success_criteria`, `expected_value`, `risks`. Also capture `poc_goal` (what the first proof of concept should demonstrate) — this is a wizard addition not in §8.1 but required for completeness checklist point 13; store it in `business_spec.poc_goal`.
  
  **Round 2 — Technical (§8.2).** `input_types`, `input_formats`, `output_types`, `language`, `domain_vocabulary` (terms / codes / controlled lists — record `"none"` if not needed), `data_sources`, `model_provider` (or `"configurable"`), `model_preferences` (model names, temperature, reasoning_effort, embedding/transcription model), `security_constraints`, `integration_targets` (record `[]` if none), `human_review` (yes / no / conditional), `evaluation_requirements` (metrics, test data, thresholds), `runtime_interface`.
  
  When discussing `output_types`, also ask

• implementation_layer/solution_wizard/README.md41 KBb43f027b951eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/scripts/check_requirements.py1.4 KB4399a759551eupstream ↗

  #!/usr/bin/env python3
  """Check requirement completeness against the Section-9 checklist (V3 Gate 1).
  
  Run before component selection. Reports which of the 13 checklist points are
  answered vs. still missing, so the wizard knows exactly what follow-up
  questions to ask. Exit code 0 when complete, 1 when points are still missing.
  
  Usage:
      python check_requirements.py --blueprint path/to/use_case.blueprint.json
  """
  
  import argparse
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.requirements import check_completeness, summary
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(
          description="Check Section-9 requirement completeness for a (draft) blueprint."
      )
      parser.add_argument("--blueprint", required=True, help="Path to use_case.blueprint.json (draft is fine)")
      args = parser.parse_args()
  
      blueprint_path = Path(args.blueprint)
      if not blueprint_path.exists():
          print(f"ERROR: Blueprint file not found: {blueprint_path}", file=sys.stderr)
          return 1
  
      try:
          blueprint = Blueprint.from_file(blueprint_path)
      except Exception as exc:
          print(f"ERROR: Failed to load blueprint: {exc}", file=sys.stderr)
          return 1
  
      print(summary(blueprint))
      return 0 if not check_completeness(blueprint) else 1
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/generate_bpmn.py3.0 KB96bf172f74bdupstream ↗

  #!/usr/bin/env python3
  """Generate a BPMN 2.0 diagram from a blueprint and write workflow.bpmn.
  
  The BPMN is the *visual blueprint*: a standards-based, Level-2 business-process
  model derived from the blueprint JSON. It is linked-by-derivation -- every
  element id maps back to a blueprint object (the map is written to
  visualizations.bpmn_mapping). Keep it in sync by editing the JSON and
  regenerating; never hand-edit the diagram.
  
  Validates the blueprint before rendering by default. Use --skip-validation to
  render regardless (e.g. while iterating on a draft blueprint).
  
  Usage:
      python generate_bpmn.py --blueprint path/to/blueprint.json --output-dir path/to/output/
      python generate_bpmn.py --blueprint path/to/blueprint.json --output-dir path/to/output/ --skip-validation
  """
  
  import argparse
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.bpmn_generator import write_bpmn
  from solution_wizard.validator import validate
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(description="Generate BPMN 2.0 diagram from blueprint.")
      parser.add_argument("--blueprint", required=True, help="Path to use_case.blueprint.json")
      parser.add_argument("--output-dir", required=True, help="Directory to write workflow.bpmn")
      parser.add_argument(
          "--skip-validation",
          action="store_true",
          help="Skip blueprint validation and render anyway (useful for draft blueprints)",
      )
      args = parser.parse_args()
  
      blueprint_path = Path(args.blueprint)
      if not blueprint_path.exists():
          print(f"ERROR: Blueprint not found: {blueprint_path}", file=sys.stderr)
          return 1
  
      try:
          blueprint = Blueprint.from_file(blueprint_path)
      except Exception as exc:
          print(f"ERROR: Failed to load blueprint: {exc}", file=sys.stderr)
          return 1
  
      if not args.skip_validation:
          result = validate(blueprint)
          if not result.ok:
              print(result.summary(), file=sys.stderr)
              print(
                  "\nBlueprint has validation errors. Fix them first, or use "
                  "--skip-validation to render anyway.",
                  file=sys.stderr,
              )
              return 1
          for w in result.warnings:
              print(f"WARNING {w}")
  
      bpmn_path = write_bpmn(blueprint, args.output_dir)
      print(f"BPMN diagram written to: {bpmn_path}")
  
      # Persist the updated blueprint so visualizations.bpmn_mapping is saved.
      # write_bpmn() populates the mapping in-memory; without this the map is lost
      # when the process exits (run_wizard.py writes the blueprint, but the
      # standalone CLI must do it too to stay consistent with the documentation).
      out_blueprint = Path(args.output_dir) / "use_case.blueprint.json"
      blueprint.to_file(out_blueprint)
      print(f"Blueprint (with bpmn_mapping) saved to: {out_blueprint}")
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/generate_docs.py2.3 KB02da1d0b283dupstream ↗

  #!/usr/bin/env python3
  """Generate the use-case documentation suite (spec §18) into <output_dir>/docs/.
  
  Produces five documents from the validated blueprint: GenAI product canvas,
  technical specification, user guide, developer guide, and evaluation plan. They
  are deterministic skeletons pre-filled with blueprint facts; the wizard then
  fills the `<!-- AGENT: ... -->` narrative markers.
  
  Validates the blueprint first by default. Use --skip-validation for drafts.
  
  Usage:
      python generate_docs.py --blueprint path/to/use_case.blueprint.json --output-dir path/to/output/
  """
  
  import argparse
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.docs_generator import generate_docs
  from solution_wizard.validator import validate
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(description="Generate the use-case documentation suite.")
      parser.add_argument("--blueprint", required=True, help="Path to use_case.blueprint.json")
      parser.add_argument("--output-dir", required=True, help="Directory; docs are written to <output-dir>/docs/")
      parser.add_argument("--skip-validation", action="store_true", help="Render even if validation fails (drafts)")
      args = parser.parse_args()
  
      blueprint_path = Path(args.blueprint)
      if not blueprint_path.exists():
          print(f"ERROR: Blueprint not found: {blueprint_path}", file=sys.stderr)
          return 1
  
      try:
          blueprint = Blueprint.from_file(blueprint_path)
      except Exception as exc:
          print(f"ERROR: Failed to load blueprint: {exc}", file=sys.stderr)
          return 1
  
      if not args.skip_validation:
          result = validate(blueprint)
          if not result.ok:
              print(result.summary(), file=sys.stderr)
              print("\nBlueprint has validation errors. Fix them first, or use --skip-validation.", file=sys.stderr)
              return 1
          for w in result.warnings:
              print(f"WARNING {w}")
  
      written = generate_docs(blueprint, args.output_dir)
      print("Documentation suite written:")
      for p in written:
          print(f"  - {p}")
      print("\nNext: fill the <!-- AGENT: ... --> narrative sections in each document.")
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/generate_mermaid.py2.2 KB9835e0135f1aupstream ↗

  #!/usr/bin/env python3
  """Generate a Mermaid diagram from a blueprint and write workflow.mmd.
  
  Validates the blueprint before rendering by default. Use --skip-validation
  to render regardless (e.g. while iterating on a draft blueprint).
  
  Usage:
      python generate_mermaid.py --blueprint path/to/blueprint.json --output-dir path/to/output/
      python generate_mermaid.py --blueprint path/to/blueprint.json --output-dir path/to/output/ --skip-validation
  """
  
  import argparse
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.validator import validate
  from solution_wizard.visualizer import write_mermaid
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(description="Generate Mermaid diagram from blueprint.")
      parser.add_argument("--blueprint", required=True, help="Path to use_case.blueprint.json")
      parser.add_argument("--output-dir", required=True, help="Directory to write workflow.mmd")
      parser.add_argument(
          "--skip-validation",
          action="store_true",
          help="Skip blueprint validation and render anyway (useful for draft blueprints)",
      )
      args = parser.parse_args()
  
      blueprint_path = Path(args.blueprint)
      if not blueprint_path.exists():
          print(f"ERROR: Blueprint not found: {blueprint_path}", file=sys.stderr)
          return 1
  
      try:
          blueprint = Blueprint.from_file(blueprint_path)
      except Exception as exc:
          print(f"ERROR: Failed to load blueprint: {exc}", file=sys.stderr)
          return 1
  
      if not args.skip_validation:
          result = validate(blueprint)
          if not result.ok:
              print(result.summary(), file=sys.stderr)
              print(
                  "\nBlueprint has validation errors. Fix them first, or use "
                  "--skip-validation to render anyway.",
                  file=sys.stderr,
              )
              return 1
          for w in result.warnings:
              print(f"WARNING {w}")
  
      mmd_path = write_mermaid(blueprint, args.output_dir)
      print(f"Mermaid diagram written to: {mmd_path}")
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/generate_schema.py14 KBa0b476ae39eaupstream ↗

  #!/usr/bin/env python3
  """Generate the extraction schema using the GAIK SchemaGenerator.
  
  Called once during Phase 5 (Schema Design) of the wizard conversation.
  The wizard presents the generated schema to the user for review; the user
  approves or requests changes; approved changes are incorporated by the
  wizard directly (no second SchemaGenerator call). The approved schema is
  then saved and reused by run_poc.py via load_schema().
  
  Flow:
      Phase 5 in SKILL.md:
          1. Call this script to generate the schema from extraction_requirements.md
          2. Present output_schema.py to the user for review
          3. User approves or requests changes (wizard edits directly, no re-generation)
          4. Confirmed schema is already saved -- scaffold_poc.py will use it as-is
  
  Usage:
      python scripts/generate_schema.py \\
          --requirements poc/prompts/extraction_requirements.md \\
          --schema-name MaintenanceTicket \\
          --output-dir poc/
          [--provider azure|openai|google|aitta|litellm|...]
          [--use-azure / --no-azure]
          [--model gpt-6-luna]
          [--base-url https://your-inference-server.example/v1]
  
  An explicit --provider overrides the legacy Azure flags. Without --provider,
  Azure remains the default and --no-azure selects OpenAI. Credentials come from
  the selected provider's environment variables, never command-line arguments.
  """
  
  from __future__ import annotations
  
  import argparse
  import json
  import sys
  from pathlib import Path
  from typing import Literal, Union, get_args, get_origin
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  # GAIK imports -- only needed at runtime, not at scaffold time
  try:
      from gaik.software_components.extractor.schema import (
          DECIMAL_PERSISTED_HELPER_SOURCE,
          CompositeExtractionRequirements,
          ExtractionRequirements,
          SchemaGenerator,
          decimal_field_repr,
      )
      from gaik.software_components.llm import Provider, get_llm_config
  
      _GAIK_AVAILABLE = True
  except ImportError as _import_err:
      _GAIK_AVAILABLE = False
      _IMPORT_ERR = _import_err
  
  
  def _annotation_repr(annotation) -> str:
      """Return a Python source representation for common Pydantic field types.
  
      Routes Decimal fields to the safe DecimalField/OptionalDecimalField
      aliases instead of the plain type -- see decimal_field_repr's docstring
      (gaik.software_components.extractor.schema) for why field.annotation
      alone can no longer tell a "made safe" Decimal field apart from a plain
      one, and _DECIMAL_HELPER_SOURCE below for what those aliases need.
      """
      decimal_repr = decimal_field_repr(annotation)
      if decimal_repr is not None:
          return decimal_repr
  
      origin = get_origin(annotation)
  
      if origin is list:
          args = get_args(annotation)
          return f"list[{_annotation_repr(args[0])}]" if args else "list"
  
      if origin is Literal:
          args = get_args(annotation)
          return f"Literal[{', '.join(repr(arg) for arg in args)}]"
  
      if origin is Union:
          args = get_args(annotation)
          non_none = [arg for arg in args if arg is not type(None)]
          if len(non_none) == 1 and type(None) in args:
              return f"Optional[{_annotation_repr(non_none[0])}]"
          return f"Union[{', '.join(_annotation_repr(arg) for arg in args)}]"
  
      # Python 3.10+ union syntax, e.g. str | None.
      try:
          import types as _types
  
          if isinstance(annotation, _types.UnionType):
              args = get_args(annotation)
              non_none = [arg for arg in args if arg is not type(None)]
              if len(non_none) == 1 and type(None) in args:
                  return f"Optional[{_annotation_repr(non_none[0])}]"
              return f"Union[{', '.join(_annotation_repr(arg) for arg in args)}]"
      except AttributeError:
          pass
  
      if annotation is type(None):
          return "None"
  
      if hasattr(annotation, "__name__"):
          return annotation.__name__
  
      return repr(annotation)
  
  
  def _collect_models(model: type) -> list[type]:
      """Collect nested Pydantic models before the parent model."""
      seen: set[type] = set()
      ordered: list[type] = []
  
      def collect(current: type) -> None:
          if current in seen:
              return
          seen.add(current)
          for field in current.model_fields.values():
              if get_origin(field.annotation) is list:
                  args = get_args(field.annotation)
                  if args and isinstance(args[0], type) and hasattr(args[0], "model_fields"):
                      collect(args[0])
          ordered.append(current)
  
      collect(model)
      return ordered
  
  
  def _schema_to_py(schema_class: type, schema_name: str) -> str:
      """Render the generated Pydantic model (and any nested child models) to a
      .py file string.
  
      Builds source directly from model_fields rather than via
      print_pydantic_schema(): that helper reads field.annotation, which
      strips the Annotated/WithJsonSchema/BeforeValidator wrapping
      create_extraction_model() applies to Decimal fields (see
      decimal_field_repr's docstring) -- regenerating from the bare
      annotation would silently drop the safe Decimal representation,
      restoring both the regex-lookaround schema-rejection failure and the
      "12.40 EUR"-crashes-Decimal-parsing failure on the next PoC run.
      """
      class_blocks: list[str] = []
      needs_decimal_helper = False
  
      for current_model in _collect_models(schema_class):
          lines = [f"class {current_model.__name__}(BaseModel):"]
  
          docstring = (current_model.__doc__ or "").strip()
          if docstring:
              lines.append(f'    """{docstring}"""')
  
          lines.append("    model_config = ConfigDict(extra='forbid')")
          lines.append("")
  
          for field_name, field in current_model.model_fields.items():
              field_args: list[str] = []
  
              if field.description:
                  field_args.append(f"description={field.description!r}")
  
              if field.default_factory is not None:
                  factory_name = getattr(field.default_factory, "__name__", None)
                  if factory_name in {"list", "dict", "set"}:
                      field_args.append(f"default_factory={factory_name}")
                  else:
                      field_args.append(f"default={field.default_factory()!r}")
              elif not field.is_required():
                  field_args.append(f"default={field.default!r}")
  
              if decimal_field_repr(field.annotation) is not None:
                  needs_decimal_helper = True
              annotation = _annotation_repr(field.annotation)
              if field_args:
                  lines.append(f"    {field_name}: {annotation} = Field({', '.join(field_args)})")
              else:
                  lines.append(f"    {field_name}: {annotation}")
  
          class_blocks.append("\n".join(lines))
  
      header = (
          '"""Output schema for this use case.\n\n'
          "Generated by GAIK SchemaGenerator via the Solution Configuration Wizard.\n"
          "Reviewed and approved by the user during Phase 5 (Schema Design).\n"
          'Do not hand-edit -- re-run generate_schema.py if the requirements change.\n"""\n\n'
          "from decimal import Decimal\n"
          "from typing import List, Literal, Optional, Union\n\n"
          "from pydantic import BaseModel, Field, ConfigDict\n\n"
      )
      if needs_decimal_helper:
          header += DECIMAL_PERSISTED_HELPER_SOURCE.strip() + "\n\n\n"
      return header + "\n\n".join(class_blocks) + "\n"
  
  
  def _requirements_to_json(
      schema_class: type,
      requirements: ExtractionRequirements | CompositeExtractionRequirements,
  ) -> dict:
      """Produce the payload that load_schema() expects."""
      return {
          "model_name": schema_class.__name__,
          "requirements_type": (
              "parent_with_nested_list"
              if isinstance(requirements, CompositeExtractionRequirements)
              else "extraction"
          ),
          "requirements": requirements.model_dump(),
      }
  
  
  def _requirements_field_summary(
      requirements: ExtractionRequirements | CompositeExtractionRequirements,
  ) -> list[str] | dict[str, list[str]]:
      """Return printable field names for flat and repeated-record schemas."""
      if isinstance(requirements, CompositeExtractionRequirements):
          return {
              "parent": [field.field_name for field in requirements.parent_requirements.fields],
              **{
                  child.container_name: [field.field_name for field in child.requirements.fields]
                  for child in requirements.children
              },
          }
      return [field.field_name for field in requirements.fields]
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(
          description="Generate extraction schema via GAIK SchemaGenerator."
      )
      parser.add_argument(
          "--requirements",
          required=True,
          help="Path to extraction_requirements.md (user requirements text)",
      )
      parser.add_argument(
          "--schema-name",
          default=None,
          help="Expected class name (informational; SchemaGenerator decides the actual name)",
      )
      parser.add_argument(
          "--output-dir",
          required=True,
          help="PoC root directory -- schema files written to <output-dir>/schemas/",
      )
      parser.add_argument(
          "--provider",
          choices=[provider.value for provider in Provider] if _GAIK_AVAILABLE else None,
          default=None,
          help="Shared LLM provider; overrides --use-azure/--no-azure. Credentials come from env.",
      )
      parser.add_argument(
          "--use-azure",
          dest="use_azure",
          action="store_true",
          default=True,
          help="Use Azure OpenAI (default when --provider is omitted)",
      )
      parser.add_argument(
          "--no-azure",
          dest="use_azure",
          action="store_false",
          help="Use OpenAI directly when --provider is omitted",
      )
      parser.add_argument(
          "--model",
          default=None,
          help="Override the model used for schema generation",
      )
      parser.add_argument(
          "--base-url",
          default=None,
          help="Override the API base URL for compatible endpoints or LiteLLM",
      )
      args = parser.parse_args()
  
      if not _GAIK_AVAILABLE:
          print(
              f"ERROR: GAIK toolkit not available: {_IMPORT_ERR}\n"
              "Install with: pip install gaik[extract]",
              file=sys.stderr,
          )
          return 1
  
      req_path = Path(args.requirements)
      if not req_path.exists():
          print(f"ERROR: Requirements file not found: {req_path}", file=sys.stderr)
          return 1
  
      user_requirements = req_path.read_text(encoding="utf-8")
      provider = args.provider or ("azure" if args.use_azure else "openai")
      overrides = {}
      if args.model is not None:
          overrides["model"] = args.model
      if args.base_url is not None:
          overrides["base_url"] = args.base_url
      try:
          # Overrides must reach the factory before it validates required fields
          # (e.g. an OpenAI-compatible server's model and endpoint).
          config = get_llm_config(provider, **overrides)
      except ValueError as exc:
          print(f"ERROR: {exc}", file=sys.stderr)
          return 1
  
      output_dir = Path(args.output_dir).expanduser().resolve()
      schemas_dir = output_dir / "schemas"
      schemas_dir.mkdir(parents=True, exist_ok=True)
  
      print(f"Generating schema from: {req_path}")
      print(f"Output directory:       {schemas_dir}")
      print(f"Provider:               {provider}")
      print(f"Model:                  {config['model']}")
      print()
  
      # -- Call SchemaGenerator --
      generator = SchemaGenerator(config=config)
      schema_class = generator.generate_schema(user_requirements=user_requirements)
      requirements: ExtractionRequirements | CompositeExtractionRequirements = (
          generator.item_requirements
      )
  
      # -- Write output_schema.py --
      py_content = _schema_to_py(schema_class, args.schema_name or schema_class.__name__)
      py_path = schemas_dir / "output_schema.py"
      py_path.write_text(py_content, encoding="utf-8")
      print(f"\nSchema written to:       {py_path}")
  
      # -- Write output_schema_requirements.json (load_schema() contract) --
      req_payload = _requirements_to_json(schema_class, requirements)
      req_path_out = schemas_dir / "output_schema_requirements.json"
      req_path_out.write_text(json.dumps(req_payload, indent=2), encoding="utf-8")
      print(f"Requirements written to: {req_path_out}")
  
      # -- Export JSON Schema (best-effort, for documentation) --
      try:
          json_schema = schema_class.model_json_schema()
          json_path = schemas_dir / "output_schema.json"
          json_path.write_text(json.dumps(json_schema, indent=2), encoding="utf-8")
          print(f"JSON Schema written to:  {json_path}")
      except Exception:
          pass
  
      # -- Save requirements hash so run_poc.py knows the schema is current --
      # If extraction_requirements.md is later edited, the hash mismatch will
      # trigger regeneration on the next run_poc.py invocation.
      import hashlib
  
      req_hash = hashlib.sha256(req_path.read_bytes()).hexdigest()
      # All schema files are always named output_schema.* regardless of the class
      # name. The hash file is therefore always output_schema.hash so that every
      # PoC template's _load_schema_if_fresh() finds it with schema_name="output_schema".
      hash_path = schemas_dir / "output_schema.hash"
      hash_path.write_text(req_hash)
      print(f"Requirements hash saved: {hash_path}")
  
      print(
          f"\nGenerated class name: {schema_class.__name__}"
          f"\nFields: {_requirements_field_summary(requirements)}"
          f"\n\nPresent {py_path} to the user for review."
          "\nIf the user requests changes, edit output_schema.py and"
          "\noutput_schema_requirements.json directly -- do NOT re-run this script."
      )
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/promote_template.py8.7 KB393b8d0a0181upstream ↗

  #!/usr/bin/env python3
  """Promote a validated, agent-generalised PoC into the reusable template library.
  
  "Save" is a generalize-then-save operation, NEVER a copy:
    1. The agent rewrites the validated poc/run_poc.py into a TEMPLATE -- every
       use-case-specific literal replaced with the matching ${variable}.
    2. This script VALIDATES that candidate template and, only if all checks pass,
       saves it to templates/poc/<pattern_key>/run_poc.py.tmpl plus a manifest.
  
  Promotion checks (all must pass):
    - genericity: no blueprint-specific token leaks into the template body outside ${...}
    - fills cleanly: safe_substitute leaves no unfilled ${...} for this blueprint
    - parses: the filled template is valid Python (ast.parse)
    - imports resolve: every `from gaik...import` matches a registry import_path
    - provider stages: model stages use get_stage_config (the scaffolder refuses others)
    - no duplicate: the pattern_key directory does not already exist
  
  Usage:
      python scripts/promote_template.py \
          --blueprint ~/my-use-case/use_case.blueprint.json \
          --candidate ~/my-use-case/poc/run_poc.py.tmpl \
          [--status provisional|confirmed] [--force]
  
  The --candidate file is the agent-generalised template (with ${variables}),
  NOT the concrete run_poc.py.
  """
  
  from __future__ import annotations
  
  import argparse
  import ast
  import importlib.util
  import json
  import re
  import string
  import sys
  from datetime import datetime, timezone
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.registry import get_registry
  from solution_wizard.scaffolder import _build_variables, _derive_pattern_key, _determine_pattern
  
  TEMPLATES_DIR = Path(__file__).parent.parent / "templates" / "poc"
  
  
  def _blueprint_specific_tokens(blueprint: Blueprint) -> list:
      """Tokens that must NOT appear literally in a generic template body."""
      tokens = set()
      tokens.add(blueprint.use_case.id)
      tos = blueprint.target_output_spec if isinstance(blueprint.target_output_spec, dict) else {}
      if tos.get("schema_name"):
          tokens.add(tos["schema_name"])
      for f in tos.get("fields", []):
          if len(f) >= 4:  # skip ultra-short field names that cause false positives
              tokens.add(f)
      # Concrete model names from the blueprint
      models = blueprint.models or {}
      for key in ("transcription_model", "extraction_model"):
          v = models.get(key)
          if v:
              tokens.add(v)
      # Language code -- a hardcoded "fi" would break any non-Finnish use case
      lang = (
          blueprint.technical_spec.get("language", "")
          if isinstance(blueprint.technical_spec, dict)
          else getattr(blueprint.technical_spec, "language", "")
      )
      if lang and len(lang) >= 2:
          tokens.add(lang)
      return [t for t in tokens if t]
  
  
  def _strip_placeholders(text: str) -> str:
      """Remove ${...} placeholders so the genericity scan only sees the literal body."""
      return re.sub(r"\$\{[^}]*\}", "", text)
  
  
  def _check_genericity(candidate: str, blueprint: Blueprint) -> list:
      body = _strip_placeholders(candidate)
      leaked = []
      for tok in _blueprint_specific_tokens(blueprint):
          if re.search(r"\b" + re.escape(tok) + r"\b", body):
              leaked.append(tok)
      return leaked
  
  
  def _check_fills_cleanly(candidate: str, variables: dict) -> list:
      filled = string.Template(candidate).safe_substitute(variables)
      return re.findall(r"\$\{[^}]*\}", filled)
  
  
  def _check_parses(candidate: str, variables: dict) -> str | None:
      filled = string.Template(candidate).safe_substitute(variables)
      try:
          ast.parse(filled)
          return None
      except SyntaxError as exc:
          return str(exc)
  
  
  def _check_provider_stages(candidate: str) -> bool:
      """Mirror scaffolder._determine_pattern, which refuses templates without it."""
      return "get_stage_config(" in candidate
  
  
  def _check_imports(candidate: str) -> list:
      """Return gaik imports that do not resolve to an installed module."""
      unresolved = []
      for m in re.finditer(r"from\s+(gaik[\w\.]*)\s+import", candidate):
          modpath = m.group(1)
          try:
              if importlib.util.find_spec(modpath) is None:
                  unresolved.append(modpath)
          except (ImportError, ModuleNotFoundError, ValueError):
              unresolved.append(modpath)
      return unresolved
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(description="Promote a PoC to the template library.")
      parser.add_argument("--blueprint", required=True, help="Path to the validated blueprint")
      parser.add_argument("--candidate", required=True, help="Path to the agent-generalised template (.tmpl with ${vars})")
      parser.add_argument("--status", default="provisional", choices=["provisional", "confirmed"])
      parser.add_argument("--force", action="store_true", help="Overwrite an existing template (skip duplicate check)")
      parser.add_argument("--skip-import-check", action="store_true",
                          help="Skip gaik import resolution check (use when gaik extras are not installed)")
      args = parser.parse_args()
  
      bp_path = Path(args.blueprint)
      cand_path = Path(args.candidate)
      for p, label in [(bp_path, "blueprint"), (cand_path, "candidate template")]:
          if not p.exists():
              print(f"ERROR: {label} not found: {p}", file=sys.stderr)
              return 1
  
      blueprint = Blueprint.from_file(bp_path)
      candidate = cand_path.read_text(encoding="utf-8")
      variables = _build_variables(blueprint, _determine_pattern(blueprint))
  
      pattern_key = _derive_pattern_key(blueprint)
      if pattern_key == "_generic":
          print("ERROR: cannot derive a stable pattern key (no automated steps).", file=sys.stderr)
          return 1
  
      print(f"Pattern key: {pattern_key}")
      failures = []
  
      # 1. Duplicate check
      target_dir = TEMPLATES_DIR / pattern_key
      if target_dir.exists() and not args.force:
          print(f"ERROR: template already exists for this pipeline shape: {target_dir}")
          print("This pipeline is already in the library (it would have been matched as a fixed pattern).")
          return 1
  
      # 2. Genericity scan
      leaked = _check_genericity(candidate, blueprint)
      if leaked:
          failures.append(
              f"genericity: use-case-specific tokens leaked into the template body "
              f"(must be ${{variables}}): {leaked}"
          )
  
      # 3. Fills cleanly
      unfilled = _check_fills_cleanly(candidate, variables)
      if unfilled:
          failures.append(f"fills_cleanly: unfilled placeholders for this blueprint: {sorted(set(unfilled))}")
  
      # 4. Parses
      parse_err = _check_parses(candidate, variables)
      if parse_err:
          failures.append(f"parses: filled template is not valid Python: {parse_err}")
  
      # 5. Provider stages
      if not _check_provider_stages(candidate):
          failures.append(
              "provider_stages: build each model stage with "
              'provider_config.get_stage_config(config, "<stage>"), not use_azure or a fixed model'
          )
  
      # 6. Imports resolve (optional -- needs gaik installed)
      if not args.skip_import_check:
          unresolved = _check_imports(candidate)
          if unresolved:
              failures.append(f"imports: unresolved gaik imports: {unresolved}")
  
      if failures:
          print("\nPROMOTION REJECTED:")
          for f in failures:
              print(f"  - {f}")
          print("\nFix the candidate template (or keep this PoC as a one-off) and retry.")
          return 1
  
      # All checks passed -- save template + manifest
      target_dir.mkdir(parents=True, exist_ok=True)
      (target_dir / "run_poc.py.tmpl").write_text(candidate, encoding="utf-8")
  
      components = [
          s.component for s in blueprint.workflow.steps
          if s.component and s.component not in ("LLMJudge", "custom", "human_review")
      ]
      manifest = {
          "pattern_key": pattern_key,
          "components": components,
          "topology_signature": [
              f"{s.component}({sorted(s.inputs)}->{sorted(s.outputs)})"
              for s in blueprint.workflow.steps
              if s.component and s.component not in ("LLMJudge", "custom", "human_review")
          ],
          "variables_used": sorted(set(re.findall(r"\$\{(\w+)\}", candidate))),
          "validated_by": [blueprint.use_case.id],
          "status": args.status,
          "created_at": datetime.now(timezone.utc).isoformat(),
      }
      (target_dir / "template.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
  
      print(f"\nPROMOTED. Template saved to: {target_dir / 'run_poc.py.tmpl'}")
      print(f"Manifest: {target_dir / 'template.json'}  (status={args.status})")
      print("Next time a blueprint with this pipeline shape is scaffolded, this template is reused automatically.")
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/run_wizard.py5.1 KB6cd93f36d4b3upstream ↗

  #!/usr/bin/env python3
  """GAIK Solution Configuration Wizard -- CLI entry point.
  
  In normal use the wizard runs conversationally inside Claude Code or
  Claude Desktop (via SKILL.md). This script is the standalone CLI
  equivalent: it validates a given blueprint and generates the Mermaid
  diagram, placing outputs in the user-specified directory.
  
  Usage:
      # Interactive: prompts for output directory
      python run_wizard.py
  
      # Non-interactive: validate + generate from an existing blueprint
      python run_wizard.py --blueprint path/to/blueprint.json --output-dir ~/my-use-case
  
      # Re-export JSON Schema from Pydantic models
      python run_wizard.py --export-schema
  """
  
  import argparse
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.bpmn_generator import generate_bpmn
  from solution_wizard.registry import get_registry
  from solution_wizard.validator import validate
  from solution_wizard.visualizer import write_mermaid
  
  SCHEMAS_DIR = Path(__file__).parent.parent / "schemas"
  
  
  def _prompt_output_dir() -> Path:
      print("\nGAIK Solution Configuration Wizard -- V1")
      print("=" * 50)
      while True:
          raw = input(
              "\nWhere would you like to save the generated files?\n"
              "(Enter a folder path, e.g. ~/projects/my-use-case): "
          ).strip()
          if not raw:
              print("Output directory is required.")
              continue
          output_dir = Path(raw).expanduser().resolve()
          output_dir.mkdir(parents=True, exist_ok=True)
          print(f"Output directory: {output_dir}")
          return output_dir
  
  
  def _validate_and_generate(blueprint_path: Path, output_dir: Path) -> int:
      try:
          blueprint = Blueprint.from_file(blueprint_path)
      except Exception as exc:
          print(f"ERROR: Failed to load blueprint: {exc}", file=sys.stderr)
          return 1
  
      result = validate(blueprint)
      print(result.summary())
  
      if not result.ok:
          print("\nBlueprint validation failed. Fix errors before generating outputs.")
          return 1
  
      # Generate both visual views. The BPMN is the visual blueprint, derived
      # from the JSON and linked by visualizations.bpmn_mapping (generate_bpmn
      # writes the map into the blueprint), so we render BEFORE saving the JSON
      # so the mapping is persisted alongside the blueprint.
      output_dir.mkdir(parents=True, exist_ok=True)
      mmd_path = output_dir / "workflow.mmd"
      write_mermaid(blueprint, output_dir)
      bpmn_path = output_dir / "workflow.bpmn"
      bpmn_path.write_text(generate_bpmn(blueprint), encoding="utf-8")
  
      # Save blueprint to output dir (now including visualizations.bpmn_mapping)
      out_blueprint = output_dir / "use_case.blueprint.json"
      blueprint.package.output_dir = str(output_dir)
      blueprint.to_file(out_blueprint)
      print(f"\nBlueprint saved to:  {out_blueprint}")
      print(f"Mermaid diagram:     {mmd_path}")
      print(f"BPMN diagram:        {bpmn_path}")
  
      print("\nNext steps:")
      print("  - Review the Mermaid diagram in workflow.mmd (quick flow view)")
      print("  - Open the BPMN visual blueprint workflow.bpmn in bpmn-js / Camunda / draw.io")
      print(f"  - Scaffold a runnable PoC: python scripts/scaffold_poc.py --blueprint {out_blueprint}")
      return 0
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(
          description="GAIK Solution Configuration Wizard CLI"
      )
      parser.add_argument(
          "--blueprint", help="Path to an existing use_case.blueprint.json to validate + render"
      )
      parser.add_argument(
          "--output-dir", help="Directory to save outputs (skips interactive prompt)"
      )
      parser.add_argument(
          "--export-schema",
          action="store_true",
          help="Re-export JSON Schema from Pydantic models to schemas/",
      )
      parser.add_argument(
          "--show-registry",
          action="store_true",
          help="Print a compact summary of all registry components (for component selection)",
      )
      args = parser.parse_args()
  
      if args.export_schema:
          schema_path = SCHEMAS_DIR / "use_case_blueprint.schema.json"
          Blueprint.export_json_schema(schema_path)
          print(f"JSON Schema exported to: {schema_path}")
          return 0
  
      if args.show_registry:
          print(get_registry().as_llm_context())
          return 0
  
      if args.blueprint:
          blueprint_path = Path(args.blueprint)
          if args.output_dir:
              output_dir = Path(args.output_dir).expanduser().resolve()
              output_dir.mkdir(parents=True, exist_ok=True)
          else:
              output_dir = blueprint_path.parent
          return _validate_and_generate(blueprint_path, output_dir)
  
      # Interactive mode
      output_dir = _prompt_output_dir()
      print(
          "\nTo configure a use case, describe it to the wizard in Claude Code or "
          "Claude Desktop using the /solution-wizard skill.\n"
          "Then run this script to validate and generate outputs:\n\n"
          f"  python run_wizard.py --blueprint {output_dir}/use_case.blueprint.json "
          f"--output-dir {output_dir}"
      )
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/scaffold_poc.py5.5 KB29c949a88f99upstream ↗

  #!/usr/bin/env python3
  """Scaffold a minimal runnable PoC from a validated blueprint.
  
  Validates the blueprint first, then generates the poc/ folder in the
  blueprint's output directory (or a directory you specify).
  
  Usage:
      python scaffold_poc.py --blueprint ~/my-use-case/use_case.blueprint.json
      python scaffold_poc.py --blueprint ... --output-dir ~/my-use-case
      python scaffold_poc.py --blueprint ... --synthetic        # generate sample input
      python scaffold_poc.py --blueprint ... --skip-validation  # scaffold despite errors
  """
  
  import argparse
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.scaffolder import scaffold_poc, validate_generated_python
  from solution_wizard.validator import validate
  
  # The root of the GAIK repo -- outputs must never go inside here.
  _WIZARD_ROOT = Path(__file__).parent.parent.resolve()
  _REPO_ROOT = _WIZARD_ROOT.parent.parent.resolve()
  
  
  def _check_output_dir(output_dir: Path) -> None:
      """Refuse to write inside the GAIK repo (implementation_layer/ or above).
      Exception: .wizard_workspaces paths are managed by the demo app server."""
      resolved = output_dir.resolve()
      # Allow server-managed wizard workspace directories
      if ".wizard_workspaces" in resolved.parts:
          return
      try:
          resolved.relative_to(_REPO_ROOT)
          print(
              f"ERROR: output-dir '{output_dir}' is inside the GAIK repository "
              f"({_REPO_ROOT}).\n"
              "All wizard outputs must go outside the repo. "
              "Choose a directory under your home folder or project root.",
              file=sys.stderr,
          )
          sys.exit(1)
      except ValueError:
          pass  # Not inside repo -- OK
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(description="Scaffold a PoC from a GAIK blueprint.")
      parser.add_argument("--blueprint", required=True, help="Path to use_case.blueprint.json")
      parser.add_argument(
          "--output-dir",
          help="Root directory for PoC output (defaults to blueprint.package.output_dir)",
      )
      parser.add_argument(
          "--synthetic",
          action="store_true",
          help="Generate synthetic input data in sample_input/ where possible",
      )
      parser.add_argument(
          "--skip-validation",
          action="store_true",
          help="Skip blueprint validation (scaffold anyway)",
      )
      args = parser.parse_args()
  
      blueprint_path = Path(args.blueprint)
      if not blueprint_path.exists():
          print(f"ERROR: Blueprint not found: {blueprint_path}", file=sys.stderr)
          return 1
  
      try:
          blueprint = Blueprint.from_file(blueprint_path)
      except Exception as exc:
          print(f"ERROR: Failed to load blueprint: {exc}", file=sys.stderr)
          return 1
  
      if not args.skip_validation:
          result = validate(blueprint)
          if not result.ok:
              print(result.summary(), file=sys.stderr)
              print(
                  "\nBlueprint has validation errors. Fix them or use --skip-validation.",
                  file=sys.stderr,
              )
              return 1
          for w in result.warnings:
              print(f"WARNING {w}")
  
      # Determine output directory
      if args.output_dir:
          output_dir = Path(args.output_dir).expanduser().resolve()
      elif blueprint.package.output_dir:
          output_dir = Path(blueprint.package.output_dir).expanduser().resolve()
      else:
          output_dir = blueprint_path.parent.resolve()
  
      _check_output_dir(output_dir)
      output_dir.mkdir(parents=True, exist_ok=True)
  
      print(f"Scaffolding PoC into: {output_dir / 'poc'}")
      result_info = scaffold_poc(blueprint, output_dir, synthetic=args.synthetic)
  
      poc_dir = result_info["poc_dir"]
      pattern = result_info["pattern"]
      template_wired = result_info["template_wired"]
      pattern_key = result_info["pattern_key"]
      template_save_path = result_info["template_save_path"]
  
      print(f"\nPattern detected: {pattern}")
      print(f"Pipeline pattern key: {pattern_key}")
      print(f"Generated {len(result_info['files'])} file(s) in {poc_dir}")
  
      # Validate generated Python
      parse_err = validate_generated_python(poc_dir)
      if parse_err:
          print(f"WARNING: run_poc.py has a syntax issue: {parse_err}", file=sys.stderr)
      else:
          print("run_poc.py: syntax OK")
  
      if not template_wired:
          print(
              "\nNOTE: Custom/hybrid pipeline detected. run_poc.py is a per-step skeleton"
              "\nwith reference call patterns -- the wizard agent fills the wiring."
              f"\n\nIf this pipeline is validated and you choose to promote it to a reusable"
              f"\ntemplate, it would be saved to:\n  {template_save_path}"
          )
  
      # Print handoff message
      print(f"""
  Your PoC has been generated at: {poc_dir}
  
  To run it:
    1. Install dependencies:
           pip install -r {poc_dir / 'requirements.txt'}
    2. Set up your environment:
           cp {poc_dir / '.env.example'} {poc_dir / '.env'}
           -- then open .env and fill in your API key
    3. Add a sample input file:
           See {poc_dir / 'README.md'} for the expected input format
    4. Run the pipeline:
           python {poc_dir / 'run_poc.py'}
    5. Inspect the output:
           Check {poc_dir / 'output'} for the generated result
  
  When you are ready, paste the output here or describe what you see.
  The wizard will help you interpret the result and refine if needed (Gate 3).
  """)
  
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/scripts/validate_blueprint.py1.1 KBffcd64f730c7upstream ↗

  #!/usr/bin/env python3
  """Validate a blueprint JSON against all V1 rules.
  
  Usage:
      python validate_blueprint.py --blueprint path/to/use_case.blueprint.json
  """
  
  import argparse
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
  
  from solution_wizard.blueprint import Blueprint
  from solution_wizard.validator import validate
  
  
  def main() -> int:
      parser = argparse.ArgumentParser(description="Validate a GAIK use-case blueprint.")
      parser.add_argument("--blueprint", required=True, help="Path to use_case.blueprint.json")
      args = parser.parse_args()
  
      blueprint_path = Path(args.blueprint)
      if not blueprint_path.exists():
          print(f"ERROR: Blueprint file not found: {blueprint_path}", file=sys.stderr)
          return 1
  
      try:
          blueprint = Blueprint.from_file(blueprint_path)
      except Exception as exc:
          print(f"ERROR: Failed to load blueprint: {exc}", file=sys.stderr)
          return 1
  
      result = validate(blueprint)
      print(result.summary())
      return 0 if result.ok else 1
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• implementation_layer/solution_wizard/CLAUDE.md3.9 KBb6246a05ea64upstream ↗

  # Solution Wizard — Claude Code Notes
  
  ## Schema generation constraints
  
  Two constraints must be checked every time a GAIK extraction schema is generated or manually edited. Both caused runtime failures during a hospital admissions PoC (2026-06-06) and are now enforced in SKILL.md Phase 4.
  
  ### 1. ExtractionRequirements `field_type` enum
  
  When editing `output_schema_requirements.json` directly, `field_type` must be one of:
  `str` | `int` | `float` | `bool` | `list[str]` | `date` | `decimal` | `list[dict]`
  
  `"dict"` is **not** in this enum — it causes `pydantic_core.ValidationError` at runtime when `ExtractionRequirements(**data)` is called. Use:
  - `"str"` for a single nested object field
  - `"list[dict]"` for an array of objects
  
  ### 2. Azure OpenAI structured output rejects bare `dict` types
  
  When `provider: azure` (also the legacy `azure_openai` alias or Azure through LiteLLM), `output_schema.py` must never contain `dict | None` or `list[dict]` as field types. The API requires `additionalProperties: false` on every JSON object; bare Python `dict` does not satisfy this and returns HTTP 400.
  
  Replace with a named Pydantic sub-model:
  
  ```python
  class Medication(BaseModel):
      model_config = ConfigDict(extra='forbid')
      name: str | None = None
      dose: str | None = None
      frequency: str | None = None
  ```
  
  Then use `list[Medication] | None` instead of `list[dict] | None`.
  
  This applies to any field described as a nested object: medications, social history, address, line items, etc. If the GAIK SchemaGenerator emits `list[dict]` for such a field, replace it with a named sub-model before presenting the schema to the user.
  
  **Where these rules are enforced in code:**
  - `SKILL.md` Phase 4 — constraint checklist before schema approval (Step 4.6)
  - `registries/gaik_component_registry.json` — Extractor `known_limitations`
  - `src/solution_wizard/schema_designer.py` — `_TYPE_MAP` maps `dict`/`object` to `str`, not bare `dict`
  
  ## Provider configuration when wiring a PoC
  
  Use the generated `provider_config.get_stage_config(config, stage)` helper for
  `transcription`, `parser`, `extraction`, `embedding`, `answer`, and `judge`.
  It resolves the selected provider through `get_llm_config`; do not reduce all
  providers to the old `use_azure` flag or copy a credential-bearing config while
  changing its provider. Reference-card constructs assume this helper is imported.
  
  `gpt-6-luna` is the default OpenAI/Azure text-model suggestion, subject to actual
  account access and Azure deployment naming. Native provider IDs include `azure`,
  `anthropic_foundry` and `vertex`; `azure_openai` remains a legacy blueprint alias
  that the scaffolder and `get_stage_config` map to `azure` (gaik's `get_llm_config`
  rejects it). Aitta, generic OpenAI-compatible endpoints and the optional LiteLLM
  backend use the same component config contract, with model-specific capabilities.
  LiteLLM models carry their routing prefix; native model IDs do not.
  
  Keep audio on native OpenAI/Azure (or an explicitly configured existing local
  Whisper mode), and select embeddings separately from chat. Image input and
  structured output must be supported by the actual selected model. For stage
  overrides, use the constructor/config names in the reference cards; they differ
  between modules. Secrets belong only in the runtime environment, never blueprints
  or saved report configurations.
  
  For the default GPT-6 setup omit temperature unless using supported disabled
  reasoning. GPT-6 Astra does not support `reasoning_effort='none'`; Sol/Luna do.
  Custom Azure deployment names need `model_family` for family normalization.
  PostgresAgent/TabularAgent drop their `0.0` temperature default for GPT-6/GPT-5.6
  automatically; pass `temperature=None` only for other reasoning deployments that
  reject it (o-series, older gpt-5.x reasoning tiers). Follow SKILL.md Phase 6 and
  the card's current options instead of the obsolete blanket restriction against every
  model newer than GPT-5.4.
  

• implementation_layer/solution_wizard/docs/solution_wizard.md49 KB02c88385aa3dupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/examples/document_extraction_blueprint.json4.7 KB0469dd40d941upstream ↗

  {
    "blueprint_version": "1.0",
    "metadata": {
      "blueprint_id": "purchase_order_extraction_v1",
      "created_by": "solution_configuration_wizard",
      "created_at": "2026-06-04T09:00:00Z",
      "last_updated_at": "2026-06-04T09:00:00Z",
      "status": "validated"
    },
    "use_case": {
      "id": "purchase_order_extraction",
      "name": "Purchase Order Data Extraction",
      "description": "Extract structured line-item data from PDF purchase orders for ERP integration.",
      "domain": "procurement",
      "knowledge_processes": ["knowledge_capture"]
    },
    "business_spec": {
      "current_process": "Procurement staff manually re-enters PO data from PDF files into the ERP system.",
      "pain_points": [
        "Manual re-entry is slow and error-prone",
        "PO formats vary between suppliers",
        "High volume of POs creates a bottleneck"
      ],
      "intended_users": ["procurement_staff"],
      "reviewers": ["procurement_manager"],
      "expected_value": ["reduced manual effort", "faster PO processing", "fewer data entry errors"]
    },
    "technical_spec": {
      "input_types": ["pdf"],
      "output_types": ["structured_json"],
      "language": "en",
      "human_review_required": false,
      "integration_targets": ["erp_system"]
    },
    "target_output_spec": {
      "schema_name": "PurchaseOrder",
      "fields": ["po_number", "supplier_name", "order_date", "delivery_date", "line_items", "total_amount", "currency"],
      "required_fields": ["po_number", "supplier_name", "line_items", "total_amount"]
    },
    "assumptions": [
      {
        "id": "assumption_001",
        "text": "PO documents are primarily PDF files with text-based content, not scanned images.",
        "status": "unconfirmed",
        "impact": "component_selection",
        "recorded_by": "ComponentSelector",
        "recorded_at": "2026-06-04T09:00:00Z"
      }
    ],
    "artifacts": {
      "source_pdf": {
        "type": "pdf",
        "source": "user_upload",
        "optional": false
      },
      "purchase_order_json": {
        "type": "structured_json",
        "source": "generated",
        "optional": false,
        "final_output": true,
        "produced_by": "extract_po_fields",
        "schema_ref": "schemas/purchase_order_schema.py"
      }
    },
    "components": {
      "selected_modules": [
        {
          "id": "documents_to_structured_data",
          "name": "DocumentsToStructuredData",
          "reason": "Primary input is PDF; module handles parsing and structured extraction end-to-end."
        }
      ],
      "selected_building_blocks": ["Extractor"],
      "custom_components": []
    },
    "workflow": {
      "steps": [
        {
          "id": "upload_po",
          "name": "Upload purchase order PDF",
          "type": "user_task",
          "inputs": [],
          "outputs": ["source_pdf"],
          "parameters": {},
          "depends_on": []
        },
        {
          "id": "extract_po_fields",
          "name": "Parse and extract PO fields",
          "type": "automated_task",
          "component": "DocumentsToStructuredData",
          "inputs": ["source_pdf"],
          "outputs": ["purchase_order_json"],
          "parameters": {
            "schema_ref": "schemas/purchase_order_schema.py",
            "requirements_ref": "to_be_generated"
          },
          "depends_on": ["upload_po"]
        }
      ]
    },
    "models": {
      "provider": "azure_openai",
      "extraction_model": "gpt-5.4",
      "temperature": 0.0
    },
    "validation": {
      "human_review_required": false,
      "completeness_check": true,
      "hallucination_check": false
    },
    "evaluation": {
      "metrics": ["field_exact_match", "semantic_match", "completeness"],
      "eval_framework": "extraction_eval",
      "sample_data_dir": "data/samples",
      "ground_truth_dir": "evals/ground_truth"
    },
    "governance": {
      "data_handling": {
        "external_model_api_allowed": true,
        "local_processing_required": false,
        "stores_input_data": false,
        "stores_outputs": true,
        "retention_policy": "not_specified",
        "contains_personal_data": "no",
        "output_sensitivity": "medium",
        "audit_log_required": false
      }
    },
    "runtime": {
      "interface": "cli",
      "entrypoint": "src/purchase_order_extraction/main.py",
      "env_file": ".env.example"
    },
    "package": {
      "name": "purchase_order_extraction_use_case",
      "output_dir": "/path/provided/by/user",
      "include_tests": true,
      "include_evals": true,
      "include_docs": true
    },
    "visualizations": {},
    "traceability": [
      {"component": "DocumentsToStructuredData", "required_by": "input_type=pdf"},
      {"component": "Extractor", "required_by": "output_type=structured_json; field extraction required"}
    ],
    "change_log": [
      {
        "version": "1.0",
        "date": "2026-06-04",
        "change": "Initial blueprint generated by solution_configuration_wizard V1",
        "changed_by": "solution_configuration_wizard"
      }
    ]
  }
  

• implementation_layer/solution_wizard/examples/incident_reporting_blueprint.json7.6 KBe991327b29e4upstream ↗

  {
    "blueprint_version": "1.0",
    "metadata": {
      "blueprint_id": "incident_reporting_v1",
      "created_by": "solution_configuration_wizard",
      "created_at": "2026-06-04T09:00:00Z",
      "last_updated_at": "2026-06-04T09:00:00Z",
      "status": "validated"
    },
    "use_case": {
      "id": "incident_reporting",
      "name": "Incident and Safety Observation Reporting Assistant",
      "description": "Convert spoken incident observations into structured incident reports for review and submission.",
      "domain": "manufacturing_safety",
      "knowledge_processes": ["knowledge_capture"]
    },
    "business_spec": {
      "current_process": "Employees report incidents by accessing a computer and filling a web form.",
      "pain_points": [
        "Reporting may be delayed",
        "The form is not convenient in shop-floor or outdoor contexts",
        "Report quality varies between employees"
      ],
      "intended_users": ["employee_reporter"],
      "reviewers": ["safety_manager"],
      "expected_value": ["faster reporting", "more consistent documentation", "improved safety transparency"]
    },
    "technical_spec": {
      "input_types": ["audio"],
      "output_types": ["structured_json"],
      "language": "fi",
      "human_review_required": true,
      "integration_targets": ["incident_reporting_database_optional"]
    },
    "target_output_spec": {
      "schema_name": "IncidentReport",
      "fields": ["incident_date", "location", "description", "severity", "injured_persons", "actions_taken", "reporter_name"],
      "required_fields": ["incident_date", "location", "description", "severity"]
    },
    "assumptions": [
      {
        "id": "assumption_001",
        "text": "Audio recordings are in Finnish and typically 1-5 minutes in length.",
        "status": "unconfirmed",
        "impact": "component_selection",
        "recorded_by": "ComponentSelector",
        "recorded_at": "2026-06-04T09:00:00Z"
      },
      {
        "id": "assumption_002",
        "text": "Human review is required before the report is submitted to the safety database.",
        "status": "confirmed",
        "impact": "workflow_design",
        "recorded_by": "SpecificationBuilder",
        "recorded_at": "2026-06-04T09:01:00Z"
      }
    ],
    "artifacts": {
      "incident_audio": {
        "type": "audio",
        "source": "user_upload",
        "optional": false
      },
      "incident_photo": {
        "type": "image",
        "source": "user_upload",
        "optional": true
      },
      "raw_transcript": {
        "type": "transcript",
        "source": "generated",
        "optional": false,
        "final_output": false,
        "produced_by": "transcribe_audio"
      },
      "enhanced_transcript": {
        "type": "enhanced_transcript",
        "source": "generated",
        "optional": false,
        "final_output": false,
        "produced_by": "enhance_transcript"
      },
      "incident_report_json": {
        "type": "structured_json",
        "source": "generated",
        "optional": false,
        "final_output": true,
        "produced_by": "extract_report_fields",
        "schema_ref": "schemas/incident_report_schema.py"
      },
      "validation_report": {
        "type": "validation_report",
        "source": "generated",
        "optional": false,
        "final_output": false,
        "produced_by": "validate_report"
      },
      "approved_report": {
        "type": "structured_json",
        "source": "generated",
        "optional": false,
        "final_output": true,
        "produced_by": "human_review"
      }
    },
    "components": {
      "selected_modules": [
        {
          "id": "audio_to_structured_data",
          "name": "AudioToStructuredData",
          "reason": "Primary input is audio; module encapsulates full transcription + extraction pipeline."
        }
      ],
      "selected_building_blocks": ["Transcriber", "TranscriptEnhancer", "Extractor", "LLMJudge"],
      "custom_components": []
    },
    "workflow": {
      "steps": [
        {
          "id": "record_observation",
          "name": "Record voice description and optional photo",
          "type": "user_task",
          "inputs": [],
          "outputs": ["incident_audio", "incident_photo"],
          "parameters": {},
          "depends_on": []
        },
        {
          "id": "transcribe_audio",
          "name": "Transcribe audio recording",
          "type": "automated_task",
          "component": "Transcriber",
          "inputs": ["incident_audio"],
          "outputs": ["raw_transcript"],
          "parameters": {"language": "fi"},
          "depends_on": ["record_observation"]
        },
        {
          "id": "enhance_transcript",
          "name": "Enhance Finnish transcript",
          "type": "automated_task",
          "component": "TranscriptEnhancer",
          "inputs": ["raw_transcript"],
          "outputs": ["enhanced_transcript"],
          "parameters": {},
          "depends_on": ["transcribe_audio"]
        },
        {
          "id": "extract_report_fields",
          "name": "Extract incident report fields",
          "type": "automated_task",
          "component": "Extractor",
          "inputs": ["enhanced_transcript"],
          "outputs": ["incident_report_json"],
          "parameters": {
            "schema_ref": "schemas/incident_report_schema.py",
            "requirements_ref": "to_be_generated"
          },
          "depends_on": ["enhance_transcript"]
        },
        {
          "id": "validate_report",
          "name": "Validate generated report",
          "type": "automated_task",
          "component": "LLMJudge",
          "inputs": ["enhanced_transcript", "incident_report_json"],
          "outputs": ["validation_report"],
          "parameters": {"rubric_ref": "to_be_generated"},
          "depends_on": ["extract_report_fields"]
        },
        {
          "id": "human_review",
          "name": "Safety manager reviews and approves report",
          "type": "human_review",
          "inputs": ["incident_report_json", "validation_report"],
          "outputs": ["approved_report"],
          "parameters": {},
          "depends_on": ["validate_report"]
        }
      ]
    },
    "models": {
      "provider": "azure_openai",
      "transcription_model": "gpt-4o-transcribe",
      "extraction_model": "gpt-5.4",
      "temperature": 0.0
    },
    "validation": {
      "human_review_required": true,
      "completeness_check": true,
      "hallucination_check": true
    },
    "evaluation": {
      "metrics": ["field_exact_match", "semantic_match", "completeness", "hallucination_rate"],
      "eval_framework": "extraction_eval",
      "sample_data_dir": "data/samples",
      "ground_truth_dir": "evals/ground_truth"
    },
    "governance": {
      "data_handling": {
        "external_model_api_allowed": true,
        "local_processing_required": false,
        "stores_input_data": false,
        "stores_outputs": true,
        "retention_policy": "not_specified",
        "contains_personal_data": "yes",
        "output_sensitivity": "high",
        "audit_log_required": true
      }
    },
    "runtime": {
      "interface": "cli",
      "entrypoint": "src/incident_reporting/main.py",
      "env_file": ".env.example"
    },
    "package": {
      "name": "incident_reporting_use_case",
      "output_dir": "/path/provided/by/user",
      "include_tests": true,
      "include_evals": true,
      "include_docs": true
    },
    "visualizations": {},
    "traceability": [
      {"component": "Transcriber", "required_by": "technical_spec.input_types contains audio"},
      {"component": "TranscriptEnhancer", "required_by": "technical_spec.language=fi; ASR error correction needed"},
      {"component": "Extractor", "required_by": "output_type=structured_json; field extraction required"},
      {"component": "LLMJudge", "required_by": "validation.hallucination_check=true; human_review=yes"}
    ],
    "change_log": [
      {
        "version": "1.0",
        "date": "2026-06-04",
        "change": "Initial blueprint generated by solution_configuration_wizard V1",
        "changed_by": "solution_configuration_wizard"
      }
    ]
  }
  

• implementation_layer/solution_wizard/examples/rag_workflow_blueprint.json5.1 KBba202a6ef174upstream ↗

  {
    "blueprint_version": "1.0",
    "metadata": {
      "blueprint_id": "internal_knowledge_qa_v1",
      "created_by": "solution_configuration_wizard",
      "created_at": "2026-06-04T09:00:00Z",
      "last_updated_at": "2026-06-04T09:00:00Z",
      "status": "validated"
    },
    "use_case": {
      "id": "internal_knowledge_qa",
      "name": "Internal Knowledge Base Q&A",
      "description": "Enable employees to ask natural-language questions over a collection of internal documents and receive cited answers.",
      "domain": "knowledge_management",
      "knowledge_processes": ["knowledge_access"]
    },
    "business_spec": {
      "current_process": "Employees search manually through shared drives and document repositories to find relevant information.",
      "pain_points": [
        "Finding the right document is time-consuming",
        "Answers require reading multiple documents",
        "Knowledge is siloed across different folders and formats"
      ],
      "intended_users": ["employee"],
      "reviewers": [],
      "expected_value": ["faster information access", "reduced time searching", "consistent answers with citations"]
    },
    "technical_spec": {
      "input_types": ["document_collection"],
      "output_types": ["answer"],
      "language": "en",
      "human_review_required": false,
      "integration_targets": []
    },
    "target_output_spec": {},
    "assumptions": [
      {
        "id": "assumption_001",
        "text": "Documents are primarily PDF and DOCX files in English.",
        "status": "unconfirmed",
        "impact": "component_selection",
        "recorded_by": "ComponentSelector",
        "recorded_at": "2026-06-04T09:00:00Z"
      },
      {
        "id": "assumption_002",
        "text": "PostgreSQL with pgvector extension is available for the vector store.",
        "status": "unconfirmed",
        "impact": "component_selection",
        "recorded_by": "ComponentSelector",
        "recorded_at": "2026-06-04T09:00:00Z"
      }
    ],
    "artifacts": {
      "document_collection": {
        "type": "document_collection",
        "source": "user_upload",
        "optional": false
      },
      "user_query": {
        "type": "text",
        "source": "user_upload",
        "optional": false
      },
      "answer_with_citations": {
        "type": "answer_with_citations",
        "source": "generated",
        "optional": false,
        "final_output": true,
        "produced_by": "run_rag"
      }
    },
    "components": {
      "selected_modules": [
        {
          "id": "rag_workflow",
          "name": "RAGWorkflow",
          "reason": "Use case is document Q&A with citations -- RAGWorkflow covers the full indexing and retrieval pipeline."
        }
      ],
      "selected_building_blocks": [],
      "custom_components": []
    },
    "workflow": {
      "steps": [
        {
          "id": "upload_documents",
          "name": "Upload document collection",
          "type": "user_task",
          "inputs": [],
          "outputs": ["document_collection"],
          "parameters": {},
          "depends_on": []
        },
        {
          "id": "ask_question",
          "name": "User submits a question",
          "type": "user_task",
          "inputs": [],
          "outputs": ["user_query"],
          "parameters": {},
          "depends_on": ["upload_documents"]
        },
        {
          "id": "run_rag",
          "name": "Index documents, retrieve context, generate cited answer",
          "type": "automated_task",
          "component": "RAGWorkflow",
          "inputs": ["document_collection", "user_query"],
          "outputs": ["answer_with_citations"],
          "parameters": {"collection_name": "internal_knowledge"},
          "depends_on": ["ask_question"]
        }
      ]
    },
    "models": {
      "provider": "azure_openai",
      "embedding_model": "text-embedding-3-large",
      "answer_model": "gpt-6-luna",
      "temperature": 0.0
    },
    "validation": {
      "human_review_required": false,
      "completeness_check": false,
      "hallucination_check": false
    },
    "evaluation": {
      "metrics": ["token_recall", "mrr", "rank_weighted_coverage"],
      "eval_framework": "RAG_eval",
      "sample_data_dir": "data/samples",
      "ground_truth_dir": "evals/ground_truth"
    },
    "governance": {
      "data_handling": {
        "external_model_api_allowed": true,
        "local_processing_required": false,
        "stores_input_data": true,
        "stores_outputs": false,
        "retention_policy": "not_specified",
        "contains_personal_data": "unknown",
        "output_sensitivity": "medium",
        "audit_log_required": false
      }
    },
    "runtime": {
      "interface": "cli",
      "entrypoint": "src/internal_knowledge_qa/main.py",
      "env_file": ".env.example"
    },
    "package": {
      "name": "internal_knowledge_qa_use_case",
      "output_dir": "/path/provided/by/user",
      "include_tests": true,
      "include_evals": true,
      "include_docs": true
    },
    "visualizations": {},
    "traceability": [
      {"component": "RAGWorkflow", "required_by": "input_type=document_collection, output_type=answer; includes its Chroma vector store"}
    ],
    "change_log": [
      {
        "version": "1.0",
        "date": "2026-06-04",
        "change": "Initial blueprint generated by solution_configuration_wizard V1",
        "changed_by": "solution_configuration_wizard"
      }
    ]
  }
  

• implementation_layer/solution_wizard/gaik_validated_version.txt7 B83ce05d72fbfupstream ↗

  0.8.11
  

• implementation_layer/solution_wizard/images/solution_wizard_v2_workflow.png1.4 MB5b12cd15d547upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/registries/component_reference_cards.json190 KB5a71abf0a425upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/registries/gaik_component_registry.json124 KB823b57d1124eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/run_wizard_interactive.py15 KBfd8e166a9f80upstream ↗

  #!/usr/bin/env python3
  """Interactive CLI session for the GAIK Solution Configuration Wizard.
  
  Starts a persistent conversation with the wizard. The wizard asks questions
  at each phase and gate; you type answers and press Enter. The session continues
  until the wizard completes (or you press Ctrl+C / Ctrl+D).
  
  --- Prerequisites -----------------------------------------------------------
  
  1. Install the wizard runner extras (once):
  
         pip install "gaik[wizard]"        # installs claude-agent-sdk + python-dotenv
  
     Or install them directly:
  
         pip install claude-agent-sdk python-dotenv
  
  2. The `claude` Code CLI must be on PATH (the SDK spawns it as a subprocess):
  
         # Verify:
         claude --version
  
     Install if missing: https://docs.anthropic.com/en/claude-code/setup
  
  --- Credentials (.env) ------------------------------------------------------
  
  The script reads credentials from ../.env (one directory above this file, i.e.
  implementation_layer/.env) by default.  Copy the template and fill in your values:
  
      cp .env.example ../.env      # then edit ../.env
  
  Required variables:
  
      CLAUDE_CODE_USE_FOUNDRY=1
      ANTHROPIC_FOUNDRY_API_KEY=<your Azure Foundry API key>
      ANTHROPIC_FOUNDRY_RESOURCE=<your Foundry resource name, e.g. my-foundry-resource>
  
  Recommended (model selection):
  
      ANTHROPIC_DEFAULT_SONNET_MODEL=claude-sonnet-5-5
  
  Optional (timeouts / telemetry):
  
      API_TIMEOUT_MS=600000
      CLAUDE_CODE_MAX_RETRIES=2
      DISABLE_TELEMETRY=1
  
  Run the preflight check to verify credentials before a full session:
  
      python run_wizard_interactive.py --preflight
  
  --- Usage -------------------------------------------------------------------
  
      # Wizard asks for the output directory interactively:
      python run_wizard_interactive.py
  
      # Pre-supply the output directory (skips that question):
      python run_wizard_interactive.py --output-dir C:\\projects\\my-use-case
  
      # Use a different .env file:
      python run_wizard_interactive.py --env-file path/to/.env
  
      # Override the model:
      python run_wizard_interactive.py --model claude-sonnet-5-5
  
  See also: run_wizard_sdk.py for a headless smoke-test (no user interaction).
  """
  
  from __future__ import annotations
  
  import argparse
  import asyncio
  import os
  import sys
  from pathlib import Path
  
  from dotenv import load_dotenv
  
  from claude_agent_sdk import (
      AssistantMessage,
      ClaudeAgentOptions,
      ClaudeSDKClient,
      ResultMessage,
      TextBlock,
      ToolResultBlock,
      ToolUseBlock,
  )
  
  WIZARD_DIR = Path(__file__).resolve().parent
  
  REQUIRED_ENV_VARS = [
      "CLAUDE_CODE_USE_FOUNDRY",
      "ANTHROPIC_FOUNDRY_API_KEY",
      "ANTHROPIC_FOUNDRY_RESOURCE",
  ]
  DEFAULT_MODEL = "claude-sonnet-5-5"
  
  # ── ANSI colours (suppressed on non-TTY / Windows without ANSI support) ──────
  
  
  def _supports_colour() -> bool:
      return sys.stdout.isatty() and os.name != "nt" or os.environ.get("FORCE_COLOR")
  
  
  RESET = "\033[0m" if _supports_colour() else ""
  BOLD = "\033[1m" if _supports_colour() else ""
  DIM = "\033[2m" if _supports_colour() else ""
  CYAN = "\033[36m" if _supports_colour() else ""
  YELLOW = "\033[33m" if _supports_colour() else ""
  GREEN = "\033[32m" if _supports_colour() else ""
  RED = "\033[31m" if _supports_colour() else ""
  
  
  # ── Environment ───────────────────────────────────────────────────────────────
  
  
  def load_environment(env_file: Path) -> dict[str, str]:
      if env_file.exists():
          load_dotenv(env_file)
      else:
          print(f"{DIM}(no .env at {env_file} — using current shell env){RESET}", file=sys.stderr)
  
      missing = [k for k in REQUIRED_ENV_VARS if not (os.getenv(k) or "").strip()]
      if missing:
          raise RuntimeError(
              f"{RED}Missing Foundry env vars: {', '.join(missing)}{RESET}\n"
              f"Copy .env.example → .env and fill them in."
          )
  
      env = dict(os.environ)
      env.setdefault("API_TIMEOUT_MS", "600000")
      env.setdefault("CLAUDE_CODE_MAX_RETRIES", "2")
      env.setdefault("DISABLE_TELEMETRY", "1")
      return env
  
  
  def resolve_model(env: dict[str, str], cli_model: str | None) -> str:
      return (
          cli_model
          or env.get("ANTHROPIC_MODEL")
          or env.get("ANTHROPIC_DEFAULT_SONNET_MODEL")
          or DEFAULT_MODEL
      ).strip()
  
  
  def guard_nested_session(allow_nested: bool, env: dict[str, str]) -> None:
      if not (os.environ.get("CLAUDECODE") or env.get("CLAUDECODE")):
          return
      if not allow_nested:
          raise RuntimeError(
              f"{RED}Detected CLAUDECODE — cannot spawn a nested Claude Code session.{RESET}\n"
              "Run from a normal terminal where CLAUDECODE is not set.\n"
              "Pass --allow-nested to override (may disrupt the parent session)."
          )
      for k in ("CLAUDECODE", "CLAUDE_CODE_ENTRYPOINT"):
          env.pop(k, None)
  
  
  # ── Output printing ───────────────────────────────────────────────────────────
  
  
  def _print_text(text: str) -> None:
      """Print wizard text in cyan so it's visually distinct from the user prompt."""
      print(f"{CYAN}{text}{RESET}", end="", flush=True)
  
  
  def _print_tool_use(name: str, inp: str) -> None:
      truncated = inp if len(inp) <= 200 else inp[:200] + " …"
      print(f"\n{DIM}[running: {name}  {truncated}]{RESET}", flush=True)
  
  
  def _print_tool_result(content: str) -> None:
      if len(content) > 600:
          content = content[:600] + "\n… [truncated]"
      print(f"{DIM}{content}{RESET}", flush=True)
  
  
  def _print_result(message: ResultMessage) -> None:
      print()  # blank line after the last streamed text
      if message.is_error:
          print(f"\n{RED}--- Session ended with error (stop: {message.stop_reason}) ---{RESET}")
          if message.result:
              print(message.result)
      else:
          print(f"\n{GREEN}--- Wizard session complete ---{RESET}")
          print(
              f"{DIM}Turns: {message.num_turns}  |  "
              f"Session: {message.session_id}"
              + (f"  |  Cost: ${message.total_cost_usd:.4f}" if message.total_cost_usd else "")
              + RESET
          )
  
  
  # ── Input ─────────────────────────────────────────────────────────────────────
  
  
  def _read_user_input() -> str | None:
      """Prompt the user and return their input, or None on EOF/Ctrl+D."""
      try:
          print(f"\n{BOLD}You:{RESET} ", end="", flush=True)
          line = sys.stdin.readline()
          if not line:  # EOF / Ctrl+D
              return None
          return line.rstrip("\n")
      except KeyboardInterrupt:
          return None
  
  
  async def _drain_response(client: ClaudeSDKClient) -> ResultMessage | None:
      """Stream one wizard turn to stdout.
  
      Returns None when the turn ended normally (wizard is waiting for the next
      user message). Returns the ResultMessage when the turn ended with an error
      (so the caller knows to stop the loop).
  
      A normal-success ResultMessage means "I finished responding — your turn."
      It does NOT mean the session is over, so we never stop the loop on success.
      """
      async for message in client.receive_response():
          if isinstance(message, AssistantMessage):
              for block in message.content:
                  if isinstance(block, TextBlock):
                      _print_text(block.text)
                  elif isinstance(block, ToolUseBlock):
                      inp = str(block.input)
                      _print_tool_use(block.name, inp)
                  elif isinstance(block, ToolResultBlock):
                      _print_tool_result(str(block.content))
          elif isinstance(message, ResultMessage):
              if message.is_error:
                  _print_result(message)
                  return message  # error — caller should stop
              # Normal turn end: wizard is waiting for user input.
              # Return None so the conversation loop continues.
              return None
      return None
  
  
  # ── Initial prompt ────────────────────────────────────────────────────────────
  
  
  def build_initial_prompt(output_dir: Path | None) -> str:
      """The first message that invokes the wizard skill and sets up the session."""
      output_hint = (
          (
              f"\nThe user has pre-selected this output directory: {output_dir}\n"
              "Skip the directory question and use it directly."
          )
          if output_dir
          else ""
      )
  
      return f"""\
  /solution-wizard
  
  You are starting an interactive GAIK Solution Configuration Wizard session.
  Load and follow the complete wizard instructions from ./SKILL.md (you are already
  in the wizard directory). {output_hint}
  
  Begin Phase 1: introduce yourself briefly, then ask the user for the output
  directory (unless already provided above) and their use-case description.
  Wait for each answer before asking the next question -- this is an interactive
  conversation, one or two questions at a time.
  """.strip()
  
  
  # ── Main interactive loop ─────────────────────────────────────────────────────
  
  
  async def run_interactive(
      env: dict[str, str],
      model: str,
      output_dir: Path | None,
  ) -> int:
      options = ClaudeAgentOptions(
          cwd=str(WIZARD_DIR),
          add_dirs=[str(output_dir)] if output_dir else [],
          env=env,
          model=model,
          allowed_tools=["Read", "Grep", "Glob", "Write", "Edit", "Bash"],
          permission_mode="bypassPermissions",
          setting_sources=[],  # isolate from repo CLAUDE.md / graphify noise
      )
  
      print(f"{BOLD}GAIK Solution Configuration Wizard{RESET}")
      print(f"{DIM}Model: {model}  |  Wizard dir: {WIZARD_DIR}")
      if output_dir:
          print(f"Output dir: {output_dir}")
      print(f"Ctrl+C or Ctrl+D to exit{RESET}\n")
  
      async with ClaudeSDKClient(options=options) as client:
          # ── First turn: invoke the skill, get the wizard's opening message ──
          initial_prompt = build_initial_prompt(output_dir)
          await client.query(initial_prompt)
          error_result = await _drain_response(client)
  
          # ── Conversation loop ────────────────────────────────────────────────
          # _drain_response returns None on a normal turn end (wizard is waiting)
          # and a ResultMessage on error. Loop until error or user exits.
          while error_result is None:
              user_input = _read_user_input()
              if user_input is None:  # EOF or Ctrl+D
                  print(f"\n{YELLOW}(session ended by user){RESET}")
                  break
              if not user_input.strip():
                  continue  # ignore blank lines
  
              await client.query(user_input)
              error_result = await _drain_response(client)
  
          if error_result is None:
              # Clean user exit — print a minimal summary.
              print(f"\n{DIM}Session ended.{RESET}")
          # Error case: _print_result was already called inside _drain_response.
  
      return 0
  
  
  # ── CLI ───────────────────────────────────────────────────────────────────────
  
  
  def parse_args() -> argparse.Namespace:
      p = argparse.ArgumentParser(
          description="Interactive CLI session for the GAIK Solution Configuration Wizard."
      )
      p.add_argument(
          "--output-dir",
          help="Pre-select the output directory (skips the wizard's directory question).",
      )
      p.add_argument(
          "--env-file",
          default=str(WIZARD_DIR.parent / ".env"),
          help="Path to .env with Foundry credentials (default: wizard dir/.env).",
      )
      p.add_argument(
          "--model",
          default=None,
          help=f"Override the model (default: ANTHROPIC_MODEL or {DEFAULT_MODEL}).",
      )
      p.add_argument(
          "--preflight",
          action="store_true",
          help="Check Foundry credentials with a tiny test query (no wizard session started).",
      )
      p.add_argument(
          "--allow-nested",
          action="store_true",
          help="Permit running inside a Claude Code session by dropping CLAUDECODE "
          "for the child process (may disrupt the parent session).",
      )
      return p.parse_args()
  
  
  def main() -> None:
      args = parse_args()
      try:
          env = load_environment(Path(args.env_file).expanduser().resolve())
          guard_nested_session(args.allow_nested, env)
          model = resolve_model(env, args.model)
  
          if args.preflight:
              from claude_agent_sdk import query, ClaudeAgentOptions
  
              print(f"Preflight: checking Foundry routing (model={model}) …\n")
  
              async def _check():
                  opts = ClaudeAgentOptions(
                      cwd=str(WIZARD_DIR),
                      env=env,
                      model=model,
                      allowed_tools=[],
                      permission_mode="bypassPermissions",
                      setting_sources=[],
                      max_turns=1,
                  )
                  async for msg in query(prompt="Reply with the single word: READY", options=opts):
                      if isinstance(msg, __import__("claude_agent_sdk").AssistantMessage):
                          for b in msg.content:
                              if hasattr(b, "text"):
                                  print(f"{GREEN}{b.text}{RESET}", end="", flush=True)
                      elif isinstance(msg, __import__("claude_agent_sdk").ResultMessage):
                          print()
                          if msg.is_error:
                              print(f"{RED}Preflight failed (stop: {msg.stop_reason}){RESET}")
                              return 1
                          cost = f"  cost ${msg.total_cost_usd:.4f}" if msg.total_cost_usd else ""
                          print(f"{GREEN}Preflight passed{RESET}{DIM}{cost}{RESET}")
                          return 0
                  return 0
  
              sys.exit(asyncio.run(_check()))
  
          output_dir: Path | None = None
          if args.output_dir:
              output_dir = Path(args.output_dir).expanduser()
              output_dir.mkdir(parents=True, exist_ok=True)
              output_dir = output_dir.resolve()
  
          sys.exit(asyncio.run(run_interactive(env, model, output_dir)))
  
      except KeyboardInterrupt:
          print(f"\n{YELLOW}Interrupted.{RESET}", file=sys.stderr)
          sys.exit(130)
      except Exception as exc:
          print(f"\n{RED}ERROR: {exc}{RESET}", file=sys.stderr)
          sys.exit(1)
  
  
  if __name__ == "__main__":
      main()
  

• implementation_layer/solution_wizard/schemas/component_registry.schema.json2.4 KB4ea785f5a574upstream ↗

  {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "title": "GAIKComponentRegistry",
    "description": "Catalog of GAIK software modules and components the wizard can select from.",
    "type": "array",
    "items": {
      "type": "object",
      "required": [
        "id",
        "name",
        "type",
        "input_artifact_types",
        "output_artifact_types",
        "required_parameters",
        "best_for",
        "known_limitations",
        "import_path",
        "source_path",
        "readme_path",
        "example_script_path"
      ],
      "properties": {
        "id": { "type": "string" },
        "name": { "type": "string" },
        "type": {
          "type": "string",
          "enum": ["software_module", "software_component"]
        },
        "input_artifact_types": {
          "type": "array",
          "items": { "type": "string" }
        },
        "output_artifact_types": {
          "type": "array",
          "items": { "type": "string" }
        },
        "knowledge_processes": {
          "type": "array",
          "items": {
            "type": "string",
            "enum": ["knowledge_capture", "knowledge_access", "knowledge_synthesis"]
          }
        },
        "uses_components": {
          "type": "array",
          "items": { "type": "string" },
          "description": "Internal building blocks for software_module entries; empty for components."
        },
        "required_parameters": {
          "type": "array",
          "items": { "type": "string" }
        },
        "optional_parameters": {
          "type": "array",
          "items": { "type": "string" }
        },
        "supported_providers": {
          "type": "array",
          "items": { "type": "string" }
        },
        "cost_relevant_parameters": {
          "type": "array",
          "items": { "type": "string" }
        },
        "quality_tradeoffs": { "type": "object" },
        "known_limitations": {
          "type": "array",
          "items": { "type": "string" }
        },
        "best_for": {
          "type": "array",
          "items": { "type": "string" }
        },
        "example_blueprint_steps": {
          "type": "array",
          "items": { "type": "object" }
        },
        "source_path": { "type": "string" },
        "import_path": { "type": "string" },
        "example_script_path": { "type": "string" },
        "readme_path": { "type": "string" },
        "install_extra": {
          "type": "string",
          "description": "pip extra for gaik[extra] in generated requirements.txt"
        }
      }
    }
  }
  

• implementation_layer/solution_wizard/schemas/use_case_blueprint.schema.json22 KB12b9d8cc029fupstream ↗

  {
    "$defs": {
      "Artifact": {
        "properties": {
          "type": {
            "title": "Type",
            "type": "string"
          },
          "source": {
            "$ref": "#/$defs/ArtifactSource"
          },
          "optional": {
            "title": "Optional",
            "type": "boolean"
          },
          "final_output": {
            "default": false,
            "title": "Final Output",
            "type": "boolean"
          },
          "externally_visible": {
            "default": false,
            "title": "Externally Visible",
            "type": "boolean"
          },
          "produced_by": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Produced By"
          },
          "schema_ref": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Schema Ref"
          },
          "overwrite": {
            "default": false,
            "title": "Overwrite",
            "type": "boolean"
          }
        },
        "required": [
          "type",
          "source",
          "optional"
        ],
        "title": "Artifact",
        "type": "object"
      },
      "ArtifactSource": {
        "enum": [
          "user_upload",
          "generated",
          "external_system",
          "static_resource"
        ],
        "title": "ArtifactSource",
        "type": "string"
      },
      "Assumption": {
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "text": {
            "title": "Text",
            "type": "string"
          },
          "status": {
            "$ref": "#/$defs/AssumptionStatus",
            "default": "unconfirmed"
          },
          "impact": {
            "default": "",
            "title": "Impact",
            "type": "string"
          },
          "recorded_by": {
            "default": "",
            "title": "Recorded By",
            "type": "string"
          },
          "recorded_at": {
            "default": "",
            "title": "Recorded At",
            "type": "string"
          }
        },
        "required": [
          "id",
          "text"
        ],
        "title": "Assumption",
        "type": "object"
      },
      "AssumptionStatus": {
        "enum": [
          "unconfirmed",
          "confirmed",
          "overridden"
        ],
        "title": "AssumptionStatus",
        "type": "string"
      },
      "BlueprintStatus": {
        "enum": [
          "draft",
          "validated",
          "poc",
          "production_ready"
        ],
        "title": "BlueprintStatus",
        "type": "string"
      },
      "BusinessProcess": {
        "properties": {
          "participants": {
            "default": [],
            "items": {
              "$ref": "#/$defs/Participant"
            },
            "title": "Participants",
            "type": "array"
          },
          "manual_steps": {
            "default": [],
            "items": {
              "$ref": "#/$defs/ManualStep"
            },
            "title": "Manual Steps",
            "type": "array"
          },
          "external_parties": {
            "default": [],
            "items": {
              "$ref": "#/$defs/ExternalParty"
            },
            "title": "External Parties",
            "type": "array"
          },
          "message_flows": {
            "default": [],
            "items": {
              "$ref": "#/$defs/MessageFlow"
            },
            "title": "Message Flows",
            "type": "array"
          },
          "exceptions": {
            "default": [],
            "items": {
              "$ref": "#/$defs/ExceptionFlow"
            },
            "title": "Exceptions",
            "type": "array"
          },
          "decision_points": {
            "default": [],
            "items": {
              "$ref": "#/$defs/DecisionPoint"
            },
            "title": "Decision Points",
            "type": "array"
          }
        },
        "title": "BusinessProcess",
        "type": "object"
      },
      "ChangeLogEntry": {
        "properties": {
          "version": {
            "title": "Version",
            "type": "string"
          },
          "date": {
            "title": "Date",
            "type": "string"
          },
          "change": {
            "title": "Change",
            "type": "string"
          },
          "changed_by": {
            "title": "Changed By",
            "type": "string"
          }
        },
        "required": [
          "version",
          "date",
          "change",
          "changed_by"
        ],
        "title": "ChangeLogEntry",
        "type": "object"
      },
      "Components": {
        "properties": {
          "selected_modules": {
            "default": [],
            "items": {
              "$ref": "#/$defs/SelectedModule"
            },
            "title": "Selected Modules",
            "type": "array"
          },
          "selected_building_blocks": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Selected Building Blocks",
            "type": "array"
          },
          "custom_components": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Custom Components",
            "type": "array"
          }
        },
        "title": "Components",
        "type": "object"
      },
      "DataHandling": {
        "properties": {
          "external_model_api_allowed": {
            "default": true,
            "title": "External Model Api Allowed",
            "type": "boolean"
          },
          "local_processing_required": {
            "default": false,
            "title": "Local Processing Required",
            "type": "boolean"
          },
          "stores_input_data": {
            "default": false,
            "title": "Stores Input Data",
            "type": "boolean"
          },
          "stores_outputs": {
            "default": true,
            "title": "Stores Outputs",
            "type": "boolean"
          },
          "retention_policy": {
            "default": "not_specified",
            "title": "Retention Policy",
            "type": "string"
          },
          "contains_personal_data": {
            "default": "unknown",
            "title": "Contains Personal Data",
            "type": "string"
          },
          "output_sensitivity": {
            "default": "unknown",
            "title": "Output Sensitivity",
            "type": "string"
          },
          "audit_log_required": {
            "default": false,
            "title": "Audit Log Required",
            "type": "boolean"
          }
        },
        "title": "DataHandling",
        "type": "object"
      },
      "DecisionBranch": {
        "properties": {
          "condition": {
            "default": "",
            "title": "Condition",
            "type": "string"
          },
          "target": {
            "default": "",
            "title": "Target",
            "type": "string"
          }
        },
        "title": "DecisionBranch",
        "type": "object"
      },
      "DecisionPoint": {
        "description": "A business gateway beyond the implicit approve/reject one.",
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "after": {
            "default": "",
            "title": "After",
            "type": "string"
          },
          "branches": {
            "default": [],
            "items": {
              "$ref": "#/$defs/DecisionBranch"
            },
            "title": "Branches",
            "type": "array"
          }
        },
        "required": [
          "id",
          "name"
        ],
        "title": "DecisionPoint",
        "type": "object"
      },
      "ExceptionFlow": {
        "description": "An alternative / error path attached to a step.",
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "attached_to": {
            "default": "",
            "title": "Attached To",
            "type": "string"
          },
          "condition": {
            "default": "",
            "title": "Condition",
            "type": "string"
          },
          "outcome": {
            "default": "end",
            "title": "Outcome",
            "type": "string"
          }
        },
        "required": [
          "id",
          "name"
        ],
        "title": "ExceptionFlow",
        "type": "object"
      },
      "ExternalParty": {
        "description": "A black-box pool outside the organisation that hands inputs in.",
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "sends": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Sends",
            "type": "array"
          },
          "to_step": {
            "default": "",
            "title": "To Step",
            "type": "string"
          }
        },
        "required": [
          "id",
          "name"
        ],
        "title": "ExternalParty",
        "type": "object"
      },
      "Governance": {
        "properties": {
          "data_handling": {
            "$ref": "#/$defs/DataHandling",
            "default": {
              "external_model_api_allowed": true,
              "local_processing_required": false,
              "stores_input_data": false,
              "stores_outputs": true,
              "retention_policy": "not_specified",
              "contains_personal_data": "unknown",
              "output_sensitivity": "unknown",
              "audit_log_required": false
            }
          }
        },
        "title": "Governance",
        "type": "object"
      },
      "KnowledgeProcess": {
        "enum": [
          "knowledge_capture",
          "knowledge_access",
          "knowledge_synthesis"
        ],
        "title": "KnowledgeProcess",
        "type": "string"
      },
      "LoopConfig": {
        "properties": {
          "max_iterations": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Max Iterations"
          },
          "loop_condition_ref": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Loop Condition Ref"
          }
        },
        "title": "LoopConfig",
        "type": "object"
      },
      "ManualStep": {
        "description": "A human business step the AI pipeline does not model as a workflow step.",
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "performed_by": {
            "default": "",
            "title": "Performed By",
            "type": "string"
          },
          "type": {
            "default": "user_task",
            "title": "Type",
            "type": "string"
          },
          "depends_on": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Depends On",
            "type": "array"
          },
          "consumes": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Consumes",
            "type": "array"
          },
          "produces": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Produces",
            "type": "array"
          }
        },
        "required": [
          "id",
          "name"
        ],
        "title": "ManualStep",
        "type": "object"
      },
      "MessageFlow": {
        "description": "A cross-pool message (dashed line in BPMN).",
        "properties": {
          "from_": {
            "default": "",
            "title": "From",
            "type": "string"
          },
          "to": {
            "default": "",
            "title": "To",
            "type": "string"
          },
          "name": {
            "default": "",
            "title": "Name",
            "type": "string"
          }
        },
        "title": "MessageFlow",
        "type": "object"
      },
      "Metadata": {
        "properties": {
          "blueprint_id": {
            "default": "",
            "title": "Blueprint Id",
            "type": "string"
          },
          "created_by": {
            "default": "solution_configuration_wizard",
            "title": "Created By",
            "type": "string"
          },
          "created_at": {
            "default": "",
            "title": "Created At",
            "type": "string"
          },
          "last_updated_at": {
            "default": "",
            "title": "Last Updated At",
            "type": "string"
          },
          "status": {
            "$ref": "#/$defs/BlueprintStatus",
            "default": "draft"
          }
        },
        "title": "Metadata",
        "type": "object"
      },
      "Package": {
        "properties": {
          "name": {
            "default": "",
            "title": "Name",
            "type": "string"
          },
          "output_dir": {
            "default": "",
            "title": "Output Dir",
            "type": "string"
          },
          "include_tests": {
            "default": true,
            "title": "Include Tests",
            "type": "boolean"
          },
          "include_evals": {
            "default": true,
            "title": "Include Evals",
            "type": "boolean"
          },
          "include_docs": {
            "default": true,
            "title": "Include Docs",
            "type": "boolean"
          }
        },
        "title": "Package",
        "type": "object"
      },
      "Participant": {
        "description": "A swimlane (human role) or system lane in the BPMN.",
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "kind": {
            "default": "human_role",
            "title": "Kind",
            "type": "string"
          },
          "default_lane_for": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Default Lane For",
            "type": "array"
          }
        },
        "required": [
          "id",
          "name"
        ],
        "title": "Participant",
        "type": "object"
      },
      "Runtime": {
        "properties": {
          "interface": {
            "default": "cli",
            "title": "Interface",
            "type": "string"
          },
          "entrypoint": {
            "default": "",
            "title": "Entrypoint",
            "type": "string"
          },
          "env_file": {
            "default": ".env.example",
            "title": "Env File",
            "type": "string"
          }
        },
        "title": "Runtime",
        "type": "object"
      },
      "SelectedModule": {
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "reason": {
            "default": "",
            "title": "Reason",
            "type": "string"
          }
        },
        "required": [
          "id",
          "name"
        ],
        "title": "SelectedModule",
        "type": "object"
      },
      "TraceabilityEntry": {
        "properties": {
          "component": {
            "title": "Component",
            "type": "string"
          },
          "required_by": {
            "title": "Required By",
            "type": "string"
          }
        },
        "required": [
          "component",
          "required_by"
        ],
        "title": "TraceabilityEntry",
        "type": "object"
      },
      "UseCase": {
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "description": {
            "title": "Description",
            "type": "string"
          },
          "domain": {
            "title": "Domain",
            "type": "string"
          },
          "knowledge_processes": {
            "default": [],
            "items": {
              "$ref": "#/$defs/KnowledgeProcess"
            },
            "title": "Knowledge Processes",
            "type": "array"
          }
        },
        "required": [
          "id",
          "name",
          "description",
          "domain"
        ],
        "title": "UseCase",
        "type": "object"
      },
      "Workflow": {
        "properties": {
          "steps": {
            "default": [],
            "items": {
              "$ref": "#/$defs/WorkflowStep"
            },
            "title": "Steps",
            "type": "array"
          }
        },
        "title": "Workflow",
        "type": "object"
      },
      "WorkflowStep": {
        "properties": {
          "id": {
            "title": "Id",
            "type": "string"
          },
          "name": {
            "title": "Name",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          },
          "component": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Component"
          },
          "inputs": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Inputs",
            "type": "array"
          },
          "outputs": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Outputs",
            "type": "array"
          },
          "parameters": {
            "additionalProperties": true,
            "default": {},
            "title": "Parameters",
            "type": "object"
          },
          "depends_on": {
            "default": [],
            "items": {
              "type": "string"
            },
            "title": "Depends On",
            "type": "array"
          },
          "loop": {
            "anyOf": [
              {
                "$ref": "#/$defs/LoopConfig"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          }
        },
        "required": [
          "id",
          "name",
          "type"
        ],
        "title": "WorkflowStep",
        "type": "object"
      }
    },
    "properties": {
      "blueprint_version": {
        "default": "1.0",
        "title": "Blueprint Version",
        "type": "string"
      },
      "metadata": {
        "$ref": "#/$defs/Metadata",
        "default": {
          "blueprint_id": "",
          "created_by": "solution_configuration_wizard",
          "created_at": "",
          "last_updated_at": "",
          "status": "draft"
        }
      },
      "use_case": {
        "$ref": "#/$defs/UseCase"
      },
      "business_spec": {
        "additionalProperties": true,
        "default": {},
        "title": "Business Spec",
        "type": "object"
      },
      "technical_spec": {
        "additionalProperties": true,
        "default": {},
        "title": "Technical Spec",
        "type": "object"
      },
      "target_output_spec": {
        "additionalProperties": true,
        "default": {},
        "title": "Target Output Spec",
        "type": "object"
      },
      "assumptions": {
        "default": [],
        "items": {
          "$ref": "#/$defs/Assumption"
        },
        "title": "Assumptions",
        "type": "array"
      },
      "artifacts": {
        "additionalProperties": {
          "$ref": "#/$defs/Artifact"
        },
        "default": {},
        "title": "Artifacts",
        "type": "object"
      },
      "inputs": {
        "additionalProperties": true,
        "default": {},
        "title": "Inputs",
        "type": "object"
      },
      "outputs": {
        "additionalProperties": true,
        "default": {},
        "title": "Outputs",
        "type": "object"
      },
      "schemas": {
        "additionalProperties": true,
        "default": {},
        "title": "Schemas",
        "type": "object"
      },
      "components": {
        "$ref": "#/$defs/Components",
        "default": {
          "selected_modules": [],
          "selected_building_blocks": [],
          "custom_components": []
        }
      },
      "workflow": {
        "$ref": "#/$defs/Workflow",
        "default": {
          "steps": []
        }
      },
      "business_process": {
        "$ref": "#/$defs/BusinessProcess",
        "default": {
          "participants": [],
          "manual_steps": [],
          "external_parties": [],
          "message_flows": [],
          "exceptions": [],
          "decision_points": []
        }
      },
      "prompts": {
        "additionalProperties": true,
        "default": {},
        "title": "Prompts",
        "type": "object"
      },
      "models": {
        "additionalProperties": true,
        "default": {},
        "title": "Models",
        "type": "object"
      },
      "validation": {
        "additionalProperties": true,
        "default": {},
        "title": "Validation",
        "type": "object"
      },
      "evaluation": {
        "additionalProperties": true,
        "default": {},
        "title": "Evaluation",
        "type": "object"
      },
      "governance": {
        "$ref": "#/$defs/Governance",
        "default": {
          "data_handling": {
            "audit_log_required": false,
            "contains_personal_data": "unknown",
            "external_model_api_allowed": true,
            "local_processing_required": false,
            "output_sensitivity": "unknown",
            "retention_policy": "not_specified",
            "stores_input_data": false,
            "stores_outputs": true
          }
        }
      },
      "runtime": {
        "$ref": "#/$defs/Runtime",
        "default": {
          "interface": "cli",
          "entrypoint": "",
          "env_file": ".env.example"
        }
      },
      "package": {
        "$ref": "#/$defs/Package",
        "default": {
          "name": "",
          "output_dir": "",
          "include_tests": true,
          "include_evals": true,
          "include_docs": true
        }
      },
      "visualizations": {
        "additionalProperties": true,
        "default": {},
        "title": "Visualizations",
        "type": "object"
      },
      "traceability": {
        "default": [],
        "items": {
          "$ref": "#/$defs/TraceabilityEntry"
        },
        "title": "Traceability",
        "type": "array"
      },
      "change_log": {
        "default": [],
        "items": {
          "$ref": "#/$defs/ChangeLogEntry"
        },
        "title": "Change Log",
        "type": "array"
      }
    },
    "required": [
      "use_case"
    ],
    "title": "Blueprint",
    "type": "object"
  }

• implementation_layer/solution_wizard/src/solution_wizard/__init__.py48 Bd369d1b2658dupstream ↗

  """GAIK Solution Configuration Wizard -- V1."""
  

• implementation_layer/solution_wizard/src/solution_wizard/blueprint.py9.7 KB99f2f8d0b0d3upstream ↗

  """Pydantic blueprint models -- authoritative schema layer for V1.
  
  These models are the single source of truth. JSON Schema in schemas/ is
  exported from them via Blueprint.model_json_schema() and must never be
  hand-edited.
  """
  
  from __future__ import annotations
  
  import json
  from enum import Enum
  from pathlib import Path
  from typing import Any, Dict, List, Optional
  
  from pydantic import BaseModel, field_validator, model_validator
  
  
  # ---------------------------------------------------------------------------
  # Enumerations
  # ---------------------------------------------------------------------------
  
  
  class KnowledgeProcess(str, Enum):
      CAPTURE = "knowledge_capture"
      ACCESS = "knowledge_access"
      SYNTHESIS = "knowledge_synthesis"
  
  
  class BlueprintStatus(str, Enum):
      DRAFT = "draft"
      VALIDATED = "validated"
      POC = "poc"
      PRODUCTION_READY = "production_ready"
  
  
  class ArtifactSource(str, Enum):
      USER_UPLOAD = "user_upload"
      GENERATED = "generated"
      EXTERNAL_SYSTEM = "external_system"
      STATIC_RESOURCE = "static_resource"
  
  
  class AssumptionStatus(str, Enum):
      UNCONFIRMED = "unconfirmed"
      CONFIRMED = "confirmed"
      OVERRIDDEN = "overridden"
  
  
  # ---------------------------------------------------------------------------
  # Sub-models
  # ---------------------------------------------------------------------------
  
  
  class Metadata(BaseModel):
      blueprint_id: str = ""
      created_by: str = "solution_configuration_wizard"
      created_at: str = ""
      last_updated_at: str = ""
      status: BlueprintStatus = BlueprintStatus.DRAFT
  
  
  class UseCase(BaseModel):
      id: str
      name: str
      description: str
      domain: str
      knowledge_processes: List[KnowledgeProcess] = []
  
  
  class Artifact(BaseModel):
      type: str
      source: ArtifactSource
      # optional has NO default -- must be explicitly set (spec §10.3)
      optional: bool
      final_output: bool = False
      externally_visible: bool = False
      produced_by: Optional[str] = None
      schema_ref: Optional[str] = None
      overwrite: bool = False
  
      @model_validator(mode="after")
      def _check_source_rules(self) -> "Artifact":
          # produced_by is for GAIK-generated intermediates only: it names the
          # workflow step whose execution creates the artifact. User-upload
          # artifacts are *captured* by a user_task step (the step lists the
          # artifact in its outputs), but they are not *produced* by the GAIK
          # pipeline -- hence produced_by must be absent for user_upload.
          # This means a user_task step can legally list a user_upload artifact
          # in step.outputs without the artifact having a produced_by field;
          # the two sides (step.outputs vs artifact.produced_by) are independent.
          if self.source == ArtifactSource.GENERATED and self.produced_by is None:
              raise ValueError("Artifact with source='generated' requires 'produced_by' to be set.")
          if self.source == ArtifactSource.USER_UPLOAD and self.produced_by is not None:
              raise ValueError("Artifact with source='user_upload' must not have 'produced_by'.")
          return self
  
  
  class LoopConfig(BaseModel):
      max_iterations: Optional[int] = None
      loop_condition_ref: Optional[str] = None
  
  
  class WorkflowStep(BaseModel):
      id: str
      name: str
      type: str  # user_task | automated_task | human_review | decision
      component: Optional[str] = None
      inputs: List[str] = []
      outputs: List[str] = []
      parameters: Dict[str, Any] = {}
      depends_on: List[str] = []
      loop: Optional[LoopConfig] = None
  
  
  class Workflow(BaseModel):
      steps: List[WorkflowStep] = []
  
  
  class SelectedModule(BaseModel):
      id: str
      name: str
      reason: str = ""
  
  
  class Components(BaseModel):
      selected_modules: List[SelectedModule] = []
      selected_building_blocks: List[str] = []
      custom_components: List[str] = []
  
  
  class Assumption(BaseModel):
      id: str
      text: str
      status: AssumptionStatus = AssumptionStatus.UNCONFIRMED
      impact: str = ""
      recorded_by: str = ""
      recorded_at: str = ""
  
  
  class DataHandling(BaseModel):
      external_model_api_allowed: bool = True
      local_processing_required: bool = False
      stores_input_data: bool = False
      stores_outputs: bool = True
      retention_policy: str = "not_specified"
      contains_personal_data: str = "unknown"
      output_sensitivity: str = "unknown"
      audit_log_required: bool = False
  
  
  class Governance(BaseModel):
      data_handling: DataHandling = DataHandling()
  
  
  class Runtime(BaseModel):
      interface: str = "cli"
      entrypoint: str = ""
      env_file: str = ".env.example"
  
  
  class Package(BaseModel):
      name: str = ""
      output_dir: str = ""
      include_tests: bool = True
      include_evals: bool = True
      include_docs: bool = True
  
  
  class TraceabilityEntry(BaseModel):
      component: str
      required_by: str
  
  
  class ChangeLogEntry(BaseModel):
      version: str
      date: str
      change: str
      changed_by: str
  
  
  # ---------------------------------------------------------------------------
  # Business process (optional Level-2 BPMN enrichment, spec §15.2)
  # ---------------------------------------------------------------------------
  #
  # This whole section is OPTIONAL. Every field defaults to empty so existing
  # blueprints (and the test suite) validate unchanged. It exists so the BPMN
  # view can be a genuine business-process model: business detail lives in the
  # JSON (derived from this section + enrichment conventions), never authored
  # only into the diagram. The BPMN is generated from here; it is never the
  # source of truth (one-directional, linked-by-derivation).
  
  
  class Participant(BaseModel):
      """A swimlane (human role) or system lane in the BPMN."""
  
      id: str
      name: str
      kind: str = "human_role"  # human_role | system
      # Step types whose tasks land in this lane when no explicit lane is given
      # (e.g. ["user_task", "human_review"]).
      default_lane_for: List[str] = []
  
  
  class ManualStep(BaseModel):
      """A human business step the AI pipeline does not model as a workflow step."""
  
      id: str
      name: str
      performed_by: str = ""  # Participant id
      type: str = "user_task"  # user_task | manual_task
      depends_on: List[str] = []  # step ids (workflow steps or other manual steps)
      consumes: List[str] = []  # artifact ids
      produces: List[str] = []  # artifact ids
  
  
  class ExternalParty(BaseModel):
      """A black-box pool outside the organisation that hands inputs in."""
  
      id: str
      name: str
      sends: List[str] = []  # artifact ids handed in via a message flow
      to_step: str = ""  # step id that receives the handoff
  
  
  class MessageFlow(BaseModel):
      """A cross-pool message (dashed line in BPMN)."""
  
      # `from`/`to` are element ids (participant id, external party id, step id,
      # or data store id). `from_` is the Python attribute (from is reserved);
      # JSON uses the natural key "from" via the alias.
      from_: str = ""
      to: str = ""
      name: str = ""
  
      model_config = {"populate_by_name": True}
  
      @model_validator(mode="before")
      @classmethod
      def _accept_from_alias(cls, data: Any) -> Any:
          if isinstance(data, dict) and "from" in data and "from_" not in data:
              data = dict(data)
              data["from_"] = data.pop("from")
          return data
  
  
  class ExceptionFlow(BaseModel):
      """An alternative / error path attached to a step."""
  
      id: str
      name: str
      attached_to: str = ""  # step id the boundary/branch hangs off
      condition: str = ""
      # "end" -> terminate; "loop_to:<step_id>" -> rework loop back to a step.
      outcome: str = "end"
  
  
  class DecisionBranch(BaseModel):
      condition: str = ""
      target: str = ""  # step id or "end"
  
  
  class DecisionPoint(BaseModel):
      """A business gateway beyond the implicit approve/reject one."""
  
      id: str
      name: str
      after: str = ""  # step id this gateway follows
      branches: List[DecisionBranch] = []
  
  
  class BusinessProcess(BaseModel):
      participants: List[Participant] = []
      manual_steps: List[ManualStep] = []
      external_parties: List[ExternalParty] = []
      message_flows: List[MessageFlow] = []
      exceptions: List[ExceptionFlow] = []
      decision_points: List[DecisionPoint] = []
  
  
  # ---------------------------------------------------------------------------
  # Top-level Blueprint
  # ---------------------------------------------------------------------------
  
  
  class Blueprint(BaseModel):
      blueprint_version: str = "1.0"
      metadata: Metadata = Metadata()
      use_case: UseCase
      business_spec: Dict[str, Any] = {}
      technical_spec: Dict[str, Any] = {}
      target_output_spec: Dict[str, Any] = {}
      assumptions: List[Assumption] = []
      artifacts: Dict[str, Artifact] = {}
      inputs: Dict[str, Any] = {}
      outputs: Dict[str, Any] = {}
      schemas: Dict[str, Any] = {}
      components: Components = Components()
      workflow: Workflow = Workflow()
      business_process: BusinessProcess = BusinessProcess()
      prompts: Dict[str, Any] = {}
      models: Dict[str, Any] = {}
      validation: Dict[str, Any] = {}
      evaluation: Dict[str, Any] = {}
      governance: Governance = Governance()
      runtime: Runtime = Runtime()
      package: Package = Package()
      visualizations: Dict[str, Any] = {}
      traceability: List[TraceabilityEntry] = []
      change_log: List[ChangeLogEntry] = []
  
      @classmethod
      def from_file(cls, path: str | Path) -> "Blueprint":
          data = json.loads(Path(path).read_text(encoding="utf-8"))
          return cls.model_validate(data)
  
      def to_file(self, path: str | Path) -> None:
          Path(path).write_text(self.model_dump_json(indent=2), encoding="utf-8")
  
      @classmethod
      def export_json_schema(cls, path: str | Path) -> None:
          """Export JSON Schema to file (regenerates schemas/use_case_blueprint.schema.json)."""
          schema = cls.model_json_schema()
          Path(path).write_text(json.dumps(schema, indent=2), encoding="utf-8")
  

• implementation_layer/solution_wizard/src/solution_wizard/bpmn_generator.py59 KB8703ca9a7b57upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/src/solution_wizard/docs_generator.py8.4 KB285180e914f0upstream ↗

  """Documentation suite generator (spec §18) -- V3.
  
  Produces the five use-case documents from a validated blueprint into
  ``<output_dir>/docs/``:
  
      genai_product_canvas.md      technical_specification.md
      user_guide.md                developer_guide.md
      evaluation_plan.md
  
  Design: deterministic skeleton + agent prose ("Python enforces structure, the
  agent authors meaning"). Each template carries ``${...}`` placeholders that this
  module fills with blueprint-derived facts, plus ``<!-- AGENT: ... -->`` markers
  the wizard fills with narrative afterwards. Mirrors ``scaffolder._write_readme``.
  
  Public API mirrors the other generators:
      generate_docs(blueprint, output_dir) -> list[Path]
  """
  
  from __future__ import annotations
  
  import string
  from pathlib import Path
  from typing import Any, Dict, List
  
  from .blueprint import Blueprint
  
  TEMPLATES_DIR = Path(__file__).parent.parent.parent / "templates" / "docs"
  
  DOC_NAMES = [
      "genai_product_canvas",
      "technical_specification",
      "user_guide",
      "developer_guide",
      "evaluation_plan",
  ]
  
  
  # ---------------------------------------------------------------------------
  # Formatting helpers
  # ---------------------------------------------------------------------------
  
  
  def _as_dict(value: Any) -> Dict[str, Any]:
      return value if isinstance(value, dict) else {}
  
  
  def _fmt_value(value: Any) -> str:
      if value is None or value == "":
          return "_not specified_"
      if isinstance(value, bool):
          return "yes" if value else "no"
      if isinstance(value, list):
          return ", ".join(str(v) for v in value) if value else "_none_"
      if isinstance(value, dict):
          return ", ".join(f"{k}: {v}" for k, v in value.items()) if value else "_none_"
      return str(value)
  
  
  def _bullets(value: Any) -> str:
      if isinstance(value, list):
          items = [str(v) for v in value if str(v).strip()]
      elif isinstance(value, dict):
          items = [f"**{k}**: {v}" for k, v in value.items()]
      elif value in (None, ""):
          items = []
      else:
          items = [str(value)]
      if not items:
          return "- _not specified_"
      return "\n".join(f"- {it}" for it in items)
  
  
  def _field(d: Dict[str, Any], key: str) -> str:
      return _fmt_value(d.get(key))
  
  
  # ---------------------------------------------------------------------------
  # Variable extraction
  # ---------------------------------------------------------------------------
  
  
  def _components_block(blueprint: Blueprint) -> str:
      comps = blueprint.components
      lines: List[str] = []
      for m in comps.selected_modules:
          mid = getattr(m, "name", None) or getattr(m, "id", "")
          reason = getattr(m, "reason", "") or ""
          lines.append(f"- **{mid}** (module){f' — {reason}' if reason else ''}")
      for b in comps.selected_building_blocks:
          lines.append(f"- **{b}** (component)")
      for c in comps.custom_components:
          lines.append(f"- **{c}** (custom)")
      return "\n".join(lines) if lines else "- _none selected_"
  
  
  def _workflow_table(blueprint: Blueprint) -> str:
      steps = blueprint.workflow.steps
      if not steps:
          return "_No workflow steps defined._"
      rows = [
          "| Step | Type | Component | Inputs | Outputs |",
          "|------|------|-----------|--------|---------|",
      ]
      for s in steps:
          comp = s.component or "—"
          params = ""
          if s.parameters:
              params = " <br/>opts: " + ", ".join(f"{k}={v}" for k, v in s.parameters.items())
          rows.append(
              f"| {s.id} | {s.type} | {comp}{params} | {', '.join(s.inputs) or '—'} | {', '.join(s.outputs) or '—'} |"
          )
      return "\n".join(rows)
  
  
  def _artifacts_block(blueprint: Blueprint) -> str:
      if not blueprint.artifacts:
          return "- _none declared_"
      lines = []
      for aid, art in blueprint.artifacts.items():
          flags = []
          if art.final_output:
              flags.append("final output")
          if art.optional:
              flags.append("optional")
          suffix = f" ({', '.join(flags)})" if flags else ""
          lines.append(f"- `{aid}` — {art.type}, source: {art.source.value}{suffix}")
      return "\n".join(lines)
  
  
  def _build_doc_variables(blueprint: Blueprint) -> Dict[str, str]:
      uc = blueprint.use_case
      bspec = _as_dict(blueprint.business_spec)
      tspec = _as_dict(blueprint.technical_spec)
      tos = _as_dict(blueprint.target_output_spec)
      dh = blueprint.governance.data_handling
  
      fields = tos.get("fields") or []
      runtime = tspec.get("runtime_interface") or "cli"
      run_cmd = (
          "python poc/run_poc.py" if "cli" in str(runtime).lower() else f"({runtime} entry point)"
      )
  
      return {
          # identity
          "use_case_name": uc.name,
          "use_case_id": uc.id,
          "domain": uc.domain or "_not specified_",
          "description": uc.description or "_not specified_",
          "knowledge_processes": _fmt_value(
              [k.value if hasattr(k, "value") else k for k in uc.knowledge_processes]
          ),
          # business
          "current_process": _field(bspec, "current_process"),
          "pain_points": _bullets(bspec.get("pain_points")),
          "proposed_solution": _field(bspec, "proposed_solution"),
          "intended_users": _fmt_value(bspec.get("intended_users")),
          "reviewers": _fmt_value(bspec.get("reviewers")),
          "stakeholders": _fmt_value(bspec.get("stakeholders")),
          "input_artifacts": _fmt_value(bspec.get("input_artifacts") or tspec.get("input_types")),
          "target_outputs": _fmt_value(bspec.get("target_outputs") or tspec.get("output_types")),
          "success_criteria": _bullets(bspec.get("success_criteria")),
          "expected_value": _bullets(bspec.get("expected_value")),
          "risks": _bullets(bspec.get("risks")),
          "poc_goal": _field(bspec, "poc_goal"),
          # technical
          "input_types": _fmt_value(tspec.get("input_types")),
          "input_formats": _fmt_value(tspec.get("input_formats")),
          "output_types": _fmt_value(tspec.get("output_types")),
          "language": _field(tspec, "language"),
          "domain_vocabulary": _fmt_value(tspec.get("domain_vocabulary")),
          "data_sources": _fmt_value(tspec.get("data_sources")),
          "model_provider": _field(tspec, "model_provider"),
          "model_preferences": _fmt_value(tspec.get("model_preferences")),
          "security_constraints": _fmt_value(tspec.get("security_constraints")),
          "integration_targets": _fmt_value(tspec.get("integration_targets")),
          "human_review": _fmt_value(tspec.get("human_review", tspec.get("human_review_required"))),
          "evaluation_requirements": _fmt_value(tspec.get("evaluation_requirements")),
          "runtime_interface": _fmt_value(runtime),
          "run_command": run_cmd,
          # target output
          "schema_name": _field(tos, "schema_name"),
          "fields_list": _bullets(fields),
          "field_count": str(len(fields)),
          "required_fields": _fmt_value(tos.get("required_fields")),
          "missing_value_policy": _field(tos, "missing_value_policy"),
          "validation_rules": _bullets(tos.get("validation_rules")),
          # governance
          "contains_personal_data": dh.contains_personal_data,
          "output_sensitivity": dh.output_sensitivity,
          "audit_log_required": "yes" if dh.audit_log_required else "no",
          # composed blocks
          "components_block": _components_block(blueprint),
          "workflow_table": _workflow_table(blueprint),
          "artifacts_block": _artifacts_block(blueprint),
      }
  
  
  # ---------------------------------------------------------------------------
  # Public API
  # ---------------------------------------------------------------------------
  
  
  def generate_docs(blueprint: Blueprint, output_dir: str | Path) -> List[Path]:
      """Generate the five documents into ``<output_dir>/docs/``.
  
      Returns the list of written paths. Templates use ``string.Template`` with
      ``safe_substitute`` so an unexpected placeholder never raises -- but the
      test suite asserts no ``${...}`` remains for the standard variable set.
      """
      variables = _build_doc_variables(blueprint)
      docs_dir = Path(output_dir) / "docs"
      docs_dir.mkdir(parents=True, exist_ok=True)
  
      written: List[Path] = []
      for name in DOC_NAMES:
          tmpl_path = TEMPLATES_DIR / f"{name}.md.tmpl"
          content = string.Template(tmpl_path.read_text(encoding="utf-8")).safe_substitute(variables)
          out_path = docs_dir / f"{name}.md"
          out_path.write_text(content, encoding="utf-8")
          written.append(out_path)
      return written
  

• implementation_layer/solution_wizard/src/solution_wizard/registry.py6.7 KB516936c5658fupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/src/solution_wizard/requirements.py8.2 KB7731a382b386upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/src/solution_wizard/scaffolder.py46 KB6828963e2448upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/src/solution_wizard/schema_designer.py19 KB9be5e8140eceupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/src/solution_wizard/selector.py3.7 KBd0a72a2513cdupstream ↗

  """Selection reference data -- transformation chains and the module-first map.
  
  The agent (via SKILL.md) owns pattern classification and component selection.
  It reads the registry fields (input_artifact_types, output_artifact_types,
  best_for, known_limitations) and reasons from them directly -- there is no
  deterministic classifier here.
  
  This module provides two pieces of *reference data* the agent consults as
  hints, not as gates:
  
    1. CHAINS / transformation_chain() -- the canonical ordered data states for
       each known pattern, so the agent structures workflows consistently.
    2. module_for_pattern() -- the module-first rule encoded as data: if a single
       GAIK module covers a pattern end-to-end, it is the preferred choice.
  
  The agent may override either hint when the use case calls for it. Pattern
  names below are conventions the agent uses in conversation; they are not a
  fixed enum the agent must match exactly.
  
  Adding a new component to the registry requires no changes here.
  """
  
  from __future__ import annotations
  
  from typing import Dict, List, Optional
  
  from .registry import Registry, get_registry
  
  
  # ---------------------------------------------------------------------------
  # Transformation chains per pattern
  # Each entry is the ordered list of data states from input to final output.
  # These are scaffolds the agent adapts -- not rigid templates.
  # ---------------------------------------------------------------------------
  
  CHAINS: Dict[str, List[str]] = {
      "audio_to_structured": [
          "audio_input",
          "raw_transcript",
          "enhanced_transcript",
          "structured_json",
          "validated_output",
          "final_output",
      ],
      "transcript_only": [
          "audio_input",
          "raw_transcript",
          "enhanced_transcript",
      ],
      "document_to_structured": [
          "document_input",
          "parsed_text",
          "structured_json",
          "validated_output",
          "final_output",
      ],
      "vision_extraction": [
          "image_input",
          "structured_json",
          "validated_output",
          "final_output",
      ],
      "rag": [
          "document_collection",
          "text_chunks",
          "vector_index",
          "retrieved_chunks",
          "answer_with_citations",
      ],
      "classification": [
          "document_input",
          "classification_result",
      ],
      "multi_source_report": [
          "source_files",
          "normalized_evidence",
          "report_sections",
          "markdown_report",
          "final_output",
      ],
      "hybrid": [
          "mixed_input",
          "intermediate_output",
          "structured_output",
      ],
  }
  
  # Maps each pattern to the single GAIK module that covers it end-to-end,
  # where one exists. This is the module-first rule encoded as data.
  _PATTERN_TO_MODULE: Dict[str, str] = {
      "audio_to_structured": "audio_to_structured_data",
      "document_to_structured": "documents_to_structured_data",
      "rag": "rag_workflow",
      "multi_source_report": "multi_source_report_generator",
  }
  
  
  def transformation_chain(pattern: str) -> List[str]:
      """Return the canonical ordered data states for a pattern (a hint).
  
      Falls back to the generic 'hybrid' chain for unknown patterns.
      """
      return CHAINS.get(pattern, CHAINS["hybrid"])
  
  
  def module_for_pattern(pattern: str, registry: Optional[Registry] = None) -> Optional[Dict]:
      """Return the registry entry for the module that covers this pattern, or None.
  
      None means no single module covers the pattern end-to-end, so the agent
      should compose a pipeline from individual components.
      """
      module_id = _PATTERN_TO_MODULE.get(pattern)
      if not module_id:
          return None
      if registry is None:
          registry = get_registry()
      return registry.lookup_by_id(module_id)
  

• implementation_layer/solution_wizard/src/solution_wizard/validator.py16 KB3d39e43fff28upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/src/solution_wizard/visualizer.py3.1 KBe7d32a544433upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/blueprint_template.json4.5 KB18116f71ae47upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/docs/developer_guide.md.tmpl1.9 KB4c4ae1d125caupstream ↗

  # Developer Guide — ${use_case_name}
  
  > Package structure, extension points, component configuration, tests, and evaluation.
  
  ## Architecture
  This use case is driven by an **executable JSON blueprint** (`use_case.blueprint.json`) — the single source of truth. The PoC, the visual workflows (`workflow.mmd`, `workflow.bpmn`), and these docs are all *generated from it*. To change behaviour, edit the blueprint and regenerate; never hand-edit generated artifacts.
  
  ## Pipeline
  ${workflow_table}
  
  ### Components and their options
  ${components_block}
  
  <!-- AGENT: for each component, note the constructor/option values set and where they live in config. -->
  
  ## Layout (PoC)
  ```
  poc/
    run_poc.py            <- pipeline entry point
    config.yaml           <- model names, temperature, paths
    requirements.txt
    .env.example
    schemas/              <- output_schema.py (Pydantic) + requirements + hash
    prompts/              <- extraction_requirements.md, validation_rubric.md
    sample_input/  output/
    evals/                <- run_basic_eval.py
  ```
  
  ## Extension points
  - **Add/replace a component:** update `components` + `workflow.steps` + `artifacts` in the blueprint, re-validate, regenerate.
  - **Change the schema/fields:** edit `target_output_spec` (and `prompts/extraction_requirements.md`); the schema regenerates on the next run via the requirements hash.
  - **Tune a component option:** set it in `workflow.steps[].parameters` (e.g. `enhanced_transcript`, `include_verification`, `hybrid_search`).
  
  ## Configuration
  - **Model provider:** ${model_provider}
  - **Model preferences:** ${model_preferences}
  - **Integration targets:** ${integration_targets}
  
  ## Tests and evaluation
  - Evaluation requirements: ${evaluation_requirements}
  - See `evals/` for the basic evaluation script and `evaluation_plan.md` for the full plan.
  
  <!-- AGENT: note any component-specific gotchas a future developer should know. -->
  

• implementation_layer/solution_wizard/templates/docs/evaluation_plan.md.tmpl1.5 KB631c7f277efeupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/docs/genai_product_canvas.md.tmpl1.3 KB990ead605c0eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/docs/technical_specification.md.tmpl1.4 KB96fc78798608upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/docs/user_guide.md.tmpl1.3 KBfd270ef64513upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/_common/config.yaml.tmpl143 B2cc0e7cf0076upstream ↗

  # Provider/stage settings. Keep credentials in .env.
  # Explicit stages override the top-level legacy provider/model fallback.
  ${settings_yaml}
  

• implementation_layer/solution_wizard/templates/poc/_common/env.example.tmpl2.3 KBe6471a0ff8d8upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/_common/pdf_report.py.tmpl8.4 KBb9f88d8100ebupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/_common/provider_config.py.tmpl2.3 KB4632197c9657upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/_common/README.md.tmpl3.1 KB6907504e849eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/_common/run_basic_eval.py.tmpl2.0 KB44756b24b377upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/_generic/run_poc.py.tmpl13 KBd83d220b20deupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/audio_to_structured/run_poc.py.tmpl8.3 KBa8a4894dda96upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/document_to_structured/run_poc.py.tmpl8.8 KBda50c73d2935upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/hybrid_dataextractor_pymupdfparser_transcriber_831947a5/run_poc.py.tmpl9.3 KB3f0d2621f2f7upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/templates/poc/hybrid_dataextractor_pymupdfparser_transcriber_831947a5/template.json599 Bfc01295069c5upstream ↗

  {
    "pattern_key": "hybrid_dataextractor_pymupdfparser_transcriber_831947a5",
    "components": [
      "Transcriber",
      "PyMuPDFParser",
      "DataExtractor"
    ],
    "topology_signature": [
      "Transcriber(['voice_note_audio']->['raw_transcript'])",
      "PyMuPDFParser(['referral_pdf']->['parsed_referral_text'])",
      "DataExtractor(['combined_clinical_text']->['patient_record_json'])"
    ],
    "variables_used": [
      "language",
      "use_case_name"
    ],
    "validated_by": [
      "clinical_documentation_extraction"
    ],
    "status": "provisional",
    "created_at": "2026-06-06T19:25:08.735044+00:00"
  }
  

• implementation_layer/solution_wizard/templates/poc/rag/run_poc.py.tmpl5.9 KB1fcf8a17406eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_audit_registry.py3.7 KBffcd773eb6aeupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_blueprint.py3.1 KB41def0ce0105upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_bpmn.py28 KB1aa6663b9559upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_decimal_schema_persistence.py13 KB468144c28344upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_docs.py4.6 KBd5abe0edff1aupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_generate_schema_cli.py7.1 KB3227d53f8023upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_hybrid_generic.py17 KBb40f88b9794bupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_mermaid.py2.2 KBd13c2fc223cdupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_pdf_report.py6.2 KBb5cf7517dcbeupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_pipeline_composite_loading.py2.5 KB73b9d754900fupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_promotion.py7.7 KBaea25d2dcafaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_provider_scaffolding.py20 KB805e98ac269cupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_reference_cards.py12 KBe9e4b8f2fbfeupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_requirements.py4.0 KBf3a52d8cd197upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_scaffolder.py8.2 KB476e31da47c5upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_schema_designer.py7.9 KB56d5e2f092f0upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_selector.py4.5 KB7ef43fa95fcaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• implementation_layer/solution_wizard/tests/test_validator.py13 KBcbaa0073d0b3upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

Plain-text copy, no login: /skill-bundles/gaik-project-gaik-toolkit-implementation-layer-solution-wizard.md