#4965 Google Antigravity /Swarm

#4965 Google Antigravity /Swarm

#4965 Tell me about Gemini /swarm

/swarm is a native multi-agent team orchestration feature (launched in preview on November 10, 2026) built into Google Antigravity alongside Gemini 4 Carbon. Instead of feeding a complex prompt into a single sequential agent loop, typing /swarm spins up an autonomous, coordinated engineering team from a single brief—complete with isolated workspaces, heterogeneous model routing, adversarial verification, and live inspection tools.

Core Architecture & Roles

Rather than relying on prompt-level instructions to keep agents from colliding, /swarm enforces structural separation across distinct roles:

• Coordinator: Typically pinned to Gemini 4 Carbon (Google’s heavy-reasoning frontier model for long-horizon planning and architecture), the coordinator decomposes the prompt into discrete sub-tasks, assigns work, and synthesizes final outputs.

• Parallel Workers: Specialized execution agents (often routed to faster, lighter models like Gemini 4 Argon or Gemini 3.8 Flash to keep costs and latency down). Crucially, workers execute inside isolated Git worktrees, preventing concurrent file edits from overwriting each other or causing merge conflicts mid-run.

• Adversarial Verifiers: Separate agents—also well-suited for Gemini 4 Carbon—whose sole mandate is to run test suites, audit diffs, and actively challenge the workers' output rather than assist in writing it.

Key Engineering & Cost Mechanisms

• Append-Only Shared Message Board: All agents communicate by posting to a shared, append-only log. This prevents the coordinator from summarizing away critical implementation details during handoffs and provides a complete, replayable audit trail of every claim, tool call, and decision.

• Zero-Token Idle Parking: Idle agents park at zero token cost while waiting on dependencies, waking up immediately when another agent posts a relevant update or request to the board.

• Heterogeneous Model Routing: You can pin specific roles to different models—using Carbon's deep thinking tokens for upfront architecture, security auditing, and verification while offloading routine file edits to Argon or Flash.

Live Observability & Human Steering

Antigravity exposes the swarm's internal state in real time via a dedicated side pane (or in the CLI via /swarm board / Alt+S):

View	Function
Board Feed	Live stream of the append-only message log showing agents debating trade-offs, reporting blockers, and posting diffs.
Members Roster	Active agent directory allowing you to inspect live tool execution or jump directly into an individual agent's trajectory to steer it mid-run.
Graph Timeline	Interactive visual replay scrubbing through how the team branched, iterated, and converged from the initial prompt to verified code.

Trade-offs & Practical Constraints

• The Coordination Tax: Even with zero-token idle parking and model hot-swapping, a parallel swarm with adversarial test re-runs consumes significantly more quota (often 10x to 50x the tokens) than a single-agent chat. Google frames /swarm economics around cost per accepted change rather than raw cost per million tokens.

• Best-Fit Workloads: Because of the orchestration overhead, /swarm shines on multi-file codebase migrations, root-cause incident investigations, architectural refactoring, and vulnerability patching—while single-file edits or quick scripts are faster and cheaper with a single Argon or Carbon session.

What are the best practices for structuring prompts, success criteria, and automated verification loops when running long-horizon tasks in Google Antigravity /swarm?

Running long-horizon tasks in Google Antigravity's /swarm requires shifting from conversational prompting to contract-driven orchestration. Because /swarm spawns parallel workers in isolated Git worktrees alongside dedicated adversarial verifiers and a shared append-only board, ambiguous prompts cause workers to duplicate effort, drift out of scope, or burn quota in unproductive loops.

1. Pre-Swarm Scoping & Prompt Structure

Never invoke /swarm directly on an underspecified problem. Antigravity's official workflow favors separating discovery/planning from parallel execution:

1. Front-load with /grill-me and /plan: If edge cases or architectural boundaries are unclear, run /grill-me first to align on constraints, then /plan to produce a reviewable specification artifact (ORIGINAL_REQUEST.md or an implementation plan) before spawning parallel workers.

2. Partition by File/Package Boundaries: /swarm performs best when the Coordinator can slice work along clean module, directory, or interface boundaries so each worker's isolated Git worktree avoids merge conflicts at integration time. Explicitly state which modules are in-scope and which directories are read-only.

3. Anchor Workspace Rules (GEMINI.md / AGENTS.md): Place universal build commands, linter flags, architecture invariants, and forbidden patterns in a root GEMINI.md or AGENTS.md file. Every spawned worker and verifier automatically ingests these rules at startup without bloating your prompt.

Recommended Prompt Blueprint:

/swarm [Objective in 1-2 sentences]

## Scope & Partitioning
- Target modules: `src/auth/`, `src/session/`, `tests/integration/`
- Read-only / Do not modify: `src/billing/`, `migrations/`
- Workstream split: Partition workers by package; share interface contracts on the board before editing callers.

## Hard Invariants & Constraints
- Preserve backward compatibility for public REST endpoints in `openapi.yaml`.
- Zero new third-party dependencies.

## Deterministic Success Criteria (Definition of Done)
1. `pytest tests/unit/auth tests/integration/` exits with code 0 (100% pass rate).
2. `ruff check src/ && mypy --strict src/` returns zero errors.
3. Verifier confirms a reproducible regression test fails on `main` and passes on the assembled candidate.

## Stop Conditions & Escalation
- Halt and post a blocker on the board if any schema change affects `src/billing/` or after 3 failed verification cycles on the same workstream.

2. Defining Machine-Verifiable Success Criteria

Because the Verifier agent is tasked with actively challenging worker outputs rather than assisting in writing code, vague criteria ("make it cleaner" or "ensure high reliability") lead to subjective debates on the message board.

• Binary Exit-Code Gates: Define completion using exact CLI commands (pytest, cargo test, npm run typecheck, linters, or custom validation scripts) that return deterministic exit codes.

• Pre-Existing Failing Tests (Red-Green Proof): For bug fixes or refactors, require a deterministic test that fails before the change and passes after. Instruct the Verifier to treat an unexecuted or skipped check as a failure.

• Combined-Candidate Re-Verification: When multiple workers modify interacting modules in separate worktrees, individual unit tests passing in isolation is insufficient. Require the Coordinator and Verifier to re-run the full integration suite on the assembled candidate and record the exact commit/revision hash in the final acceptance record.

3. Designing Automated Verification Loops

To maximize quality while controlling token burn, structure the loop around heterogeneous roles, sandboxed autonomy, and explicit circuit breakers:

Loop Component	Best Practice	Why It Matters
Role-to-Model Pinning	Pin Coordinator and Verifier to Gemini 4 Carbon; route Workers to Gemini 4 Argon or Flash.	Carbon's deep reasoning catches subtle race conditions, security flaws, and spec drift during review, while Argon executes parallel file edits cheaply.
Test-First Seeding	If test coverage is missing, have the first wave write the test harness before implementation workers start.	Prevents workers from writing tests tailored to pass their own buggy implementation.
Board Provenance	Require workers to post command outputs, diff summaries, and artifact IDs to the append-only board—never bare "done" claims.	If a worker claims success and a verifier flags a failure, the board preserves both records for auditability.
Sandbox Permissions	Configure toolPermission to proceed-in-sandbox (or auto mode) in settings.json for long runs.	Prevents workers from stalling indefinitely on routine test/build command approvals while blocking external side effects.
Quota & Retry Caps	Set explicit iteration caps (e.g., max 3 repair loops per subtask) and check remaining quota before launching.	Each worker consumes quota independently and will not auto-resume after a quota reset—manual intervention is required.

4. Mid-Run Steering & Recovery

For multi-hour runs, monitor the swarm via /swarm board (Alt+S) or the Graph view to catch stalls early:

• Unblock Tool Prompts: Use Alt+J to jump directly to any worker paused on an out-of-sandbox permission prompt.

• Fork Speculative Fixes: If a worker gets stuck in a verification loop, jump into that agent (/agents) or branch the baseline with /fork to test an alternative fix without corrupting the main worktree.

• Distill Failures into Rules: Once the swarm converges, run /learn so recurring verification failures or project-specific gotchas are written directly back into GEMINI.md for future runs.

Create a production-ready GEMINI.md rules file and a reusable /swarm prompt template tailored for a Python repository with pytest, mypy, and strict schema validation.

Below is a drop-in root GEMINI.md rules file—automatically ingested by the Coordinator, Workers, and Adversarial Verifiers when Antigravity spawns a /swarm—followed by a reusable /swarm prompt template designed for deterministic, contract-driven Python execution.

1. Root GEMINI.md (Persistent Swarm Rules & Invariants)

Save this file at the root of your repository. It establishes global architectural invariants, strict schema enforcement, and machine-verifiable exit gates so you do not have to repeat boilerplate rules in every /swarm invocation.

# GEMINI.md — Repository Architecture, Verification & Swarm Protocol

## 1. Core Engineering Invariants
- **Python Runtime & Typing:** Target Python 3.12+. Every function signature, class attribute, and module constant MUST carry complete PEP 484 type annotations. Avoid `Any` unless strictly required at an untyped external boundary, and document why inline.
- **Strict Schema Validation (Pydantic v2):**
  - All inter-module boundaries, external API payloads, CLI/config inputs, and agent state artifacts MUST be modeled using `pydantic.BaseModel` (v2).
  - Configure boundary models with `model_config = ConfigDict(extra="forbid", strict=True, frozen=True)` unless mutability is explicitly justified.
  - Never pass raw untyped `dict` objects across service or package boundaries.
- **Determinism & Auditability:**
  - Separate stochastic/LLM generation layers from deterministic validation, state transitions, and persistence layers.
  - Raise domain-specific exceptions rather than silently swallowing errors or returning sentinel `None` values on failure.
- **Dependency Hygiene:** Do NOT add third-party packages to `pyproject.toml` or `requirements.txt` unless explicitly authorized in the active `/swarm` brief.

## 2. Canonical Verification Commands (Exit Code 0 Required)
All workers and verifiers must execute verification inside the repository's active virtual environment:
- **Unit & Integration Tests:** `pytest -v --tb=short`
- **Strict Static Type Checking:** `mypy --strict src tests`
- **Lint & Format Check:** `ruff check src tests && ruff format --check src tests`

## 3. Swarm Role & Board Protocol
- **Contract-First Coordination:** Before modifying shared interfaces or Pydantic models, Workers MUST post the proposed schema diff to the shared board and obtain Coordinator sign-off so parallel Git worktrees do not diverge.
- **Evidence-Backed Completion:** Workers must never post bare "done" messages. Every completion post on the board MUST include:
  1. Modified file paths.
  2. Exact CLI command outputs and exit codes for `pytest`, `mypy --strict`, and `ruff`.
  3. Confirmation of edge-case and negative-validation test coverage (e.g., testing `ValidationError` on malformed payloads).
- **Adversarial Verifier Mandate:**
  - Treat any unexecuted test, skipped check, `# type: ignore` without an error code, or relaxed Pydantic `ConfigDict` (`extra="allow"` / `strict=False`) as a **blocking defect**.
  - For bug fixes or behavioral changes, verify a reproducible test that fails prior to the patch and passes on the integrated candidate.
  - Re-run the full verification suite on the **merged candidate** after worktree integration before marking the run complete.

## 4. Circuit Breakers & Escalation
- Halt and post a `BLOCKER` entry on the board if:
  - A required fix crosses into a directory marked `Read-Only` in the `/swarm` prompt.
  - A workstream fails the same verification gate **3 consecutive times** (do not burn quota retrying identical edits).

2. Reusable /swarm Prompt Template

Use this template when launching multi-file feature builds, refactors, or hardening passes. Fill in the bracketed fields before submitting.

/swarm [1-2 sentence high-level objective and architectural outcome]

## Context & Reference Artifacts
- Spec / Design Doc: `[path/to/spec.md or N/A]`
- Primary Schema Module: `[e.g., src/core/schemas.py]`

## Scope & Worktree Partitioning
- **In-Scope Write Paths:** `[e.g., src/pipeline/, src/validators/, tests/unit/, tests/integration/]`
- **Read-Only Paths (Do Not Modify):** `[e.g., src/legacy/, scripts/migrations/, pyproject.toml]`
- **Workstream Decomposition:**
  1. **Schema & Contract Track:** Define/update strict Pydantic v2 models (`ConfigDict(extra="forbid", strict=True)`) and custom exceptions in `[src/core/schemas.py]`; publish contract to the board first.
  2. **Implementation Track(s):** Partition parallel workers by module (`[src/module_a/]`, `[src/module_b/]`) against the frozen Pydantic contract.
  3. **Adversarial Test Track:** Write negative schema tests (`ValidationError` cases), boundary tests, and integration tests in `[tests/]`.

## Functional & Non-Functional Requirements
- [Requirement 1: e.g., Implement topological sorting for task DAGs with cycle detection raising `CyclicDependencyError`.]
- [Requirement 2: e.g., Persist validated execution artifacts to SQLite with transactional rollback on schema failure.]
- [Requirement 3: e.g., Maintain 100% backward compatibility with existing public CLI flags.]

## Deterministic Definition of Done (Verifier Gates)
The Verifier must independently execute and confirm exit code `0` on the **assembled merge candidate** for all gates below:
1. `pytest [tests/unit/ tests/integration/] -v --tb=short` (100% pass, zero skipped tests).
2. `mypy --strict [src/ tests/]` (zero errors, zero untyped defs).
3. `ruff check [src/ tests/] && ruff format --check [src/ tests/]` (zero lint or formatting violations).
4. **Schema Negative-Path Proof:** Confirm dedicated tests verify rejection of extra fields, wrong primitive types, and missing required fields across all new/modified Pydantic models.

## Model Routing & Stop Conditions
- Pin **Coordinator** and **Verifier** to `gemini-4-carbon`; route **Workers** to `gemini-4-argon`.
- Max **3 repair iterations** per workstream before escalating a `BLOCKER` to the board for human steering.

How to Deploy & Tune This Setup

1. Seed pyproject.toml to Match: Ensure [tool.mypy] (strict = true, plugins = ["pydantic.mypy"]) and [tool.pytest.ini_options] are configured in pyproject.toml so Workers and Verifiers run identical flags locally and in CI.

2. Pre-Flight Check: For large architectural refactors touching more than 5 modules, run /plan with your populated prompt first to verify the Coordinator's worktree split before releasing Workers.

How would I use this for a repeatable task: 1) identify opportunities in industry and user groups to provide innovative solutions, 2) analyze each opportunity, 3) recommend fixes and improvements, and 4) describe implementation of the fix and/or improvements?

Because /swarm excels at deterministic, code-backed execution rather than pure open-ended brainstorming, the highest-leverage way to run your 4-stage innovation pipeline repeatedly is to treat it as a schema-gated, file-backed DAG inside a dedicated repository.

Instead of asking agents to pass prose paragraphs back and forth on the message board, each stage produces a strict Pydantic v2 JSON artifact on disk (artifacts/<run_id>/stage_*.json) validated by a deterministic Python gatekeeper script (scripts/validate_stage.py) before the next stage unlocks.

1. Repository & Artifact Architecture

Set up a clean repository structure where every run is isolated by <run_id>, allowing you to rerun or branch any stage without overwriting prior work:

innovation-engine/
├── GEMINI.md                     # Permanent domain rules, scoring rubrics & gate commands
├── schemas/
│   └── pipeline.py               # Strict Pydantic v2 contracts for Stages 1–4
├── scripts/
│   ├── validate_stage.py         # Deterministic exit-code gate (schema + rubric + math checks)
│   └── render_report.py          # Compiles verified JSON artifacts into final Markdown dossier
├── prompts/
│   └── run_pipeline.md           # Parameterized /swarm prompt template
└── artifacts/
    └── <run_id>/
        ├── 01_opportunities.json # Stage 1 output
        ├── 02_analyses.json      # Stage 2 output
        ├── 03_recommendations.json # Stage 3 output
        ├── 04_implementations.json # Stage 4 output
        └── FINAL_REPORT.md       # Compiled deliverable

2. Mapping the 4 Stages to Swarm Roles & Worktrees

Stage	Swarm Execution Pattern	Model Routing	Deterministic Gate (validate_stage.py)
1. Identify Opportunities	Parallel Scout Workers partitioned by industry segment, user persona, or pain-point vector; Coordinator deduplicates and merges into 01_opportunities.json.	Workers: gemini-4-argon Coordinator: gemini-4-carbon	Enforces unique opportunity_id, non-empty evidence/citations, target user group, and measurable baseline friction metric.
2. Analyze Each Opportunity	1 Worker per opportunity_id (isolated worktrees writing 02_analysis_<id>.json, then merged into 02_analyses.json) performing root-cause, contradiction, and ICE/impact scoring.	Workers: gemini-4-argon	Verifies score math (ice_score == (impact * confidence * ease) / 10), non-empty root-cause breakdown, and filtering threshold (e.g., only top $N$ advance).
3. Recommend Fixes & Improvements	Dual-Track Workers + Adversarial Verifier: Workers propose architectural/product fixes (03_recommendations.json); Verifier challenges feasibility, hidden trade-offs, and second-order effects.	Workers: gemini-4-argon Verifier: gemini-4-carbon	Rejects hand-wavy recommendations; requires explicit mechanism of action, trade-off matrix, and falsifiable validation metric per fix.
4. Describe Implementation	Implementation Architect Workers generate phased engineering blueprints, interface specs, bills of materials, and kill-switch criteria in 04_implementations.json, then render_report.py builds FINAL_REPORT.md.	Workers: gemini-4-carbon (or argon for boilerplate)	Validates dependency ordering, resource/timeline bounds, explicit test/acceptance criteria, and compiles clean Markdown with zero broken references.

3. GEMINI.md Rules for the Innovation Pipeline

Place this in the repository root so every /swarm run automatically enforces your analytical rigor and prevents LLM fluff:

# GEMINI.md — 4-Stage Innovation Pipeline Rules

## 1. Execution & Artifact Invariants
- **No Prose-Only Handoffs:** Every stage MUST persist its structured output to `artifacts/<run_id>/` conforming strictly to `schemas/pipeline.py` (`ConfigDict(extra="forbid", strict=True)`).
- **Stage Gate Enforcement:** A worker may NOT begin Stage $N+1$ until `python scripts/validate_stage.py --run-id <run_id> --stage N` exits with code `0` and the result is posted to the swarm board.
- **Evidence & Specificity:** Never use vague marketing language ("streamline workflows", "leverage AI"). Every opportunity must cite a concrete operational bottleneck, and every fix must specify the exact physical, software, or process mechanism that resolves it.

## 2. Adversarial Verifier Rules (Stages 2–4)
- **Stage 2 Audit:** Reject any opportunity analysis that lacks a falsifiable baseline metric or inflates confidence scores without cited evidence.
- **Stage 3 Audit:** Challenge every proposed fix for unaddressed failure modes, cost/complexity regressions, or regulatory/operational friction. Force at least one revision cycle if a recommendation omits concrete trade-offs.
- **Stage 4 Audit:** Verify that implementation phases have concrete inputs, outputs, verification commands/tests, and explicit kill-switch thresholds.

4. Parameterized /swarm Prompt Template

To make this a one-command repeatable workflow, save this template in prompts/run_pipeline.md and swap out the top 4 variables for each new domain run:

/swarm Execute the 4-stage innovation discovery and implementation pipeline for run ID `<YYYY-MM-DD_slug>`.

## Run Parameters
- **Target Industry / Domain:** `[e.g., Maritime & Coastal Dock Logistics / Independent Marina Operators]`
- **Target User Groups:** `[e.g., Small marina operators, recreational vessel owners, marine service technicians]`
- **Constraints & Guardrails:** `[e.g., Capital expenditure under $5,000 per unit; zero cloud dependency required at dockside]`
- **Selection Funnel:** Identify `[8–12]` distinct opportunities in Stage 1; analyze all in Stage 2; advance only the top `[3]` highest-scoring opportunities into Stages 3 and 4.

## Stage-by-Stage Execution Plan
1. **Stage 1 (`01_opportunities.json`):** Partition Scout Workers across the target user groups to surface concrete operational bottlenecks and unmet needs. Deduplicate and validate via `python scripts/validate_stage.py --run-id <run_id> --stage 1`.
2. **Stage 2 (`02_analyses.json`):** Spawn parallel Workers (one per opportunity) to decompose root causes, underlying technical/operational contradictions, and quantitative ICE/value scores. Validate via `--stage 2` and prune to the top `[3]`.
3. **Stage 3 (`03_recommendations.json`):** Propose concrete fixes/improvements for the top `[3]` opportunities. Route through the **Adversarial Verifier** to stress-test feasibility, cost, and edge-case failure modes before running `--stage 3`.
4. **Stage 4 (`04_implementations.json` & `FINAL_REPORT.md`):** Detail phased implementation blueprints (architecture/components, step-by-step build sequence, verification criteria, and go/no-go kill switches). Validate via `--stage 4` and generate the final report with `python scripts/render_report.py --run-id <run_id>`.

## Deterministic Definition of Done
- All 4 JSON artifacts pass `validate_stage.py` with exit code `0`.
- `mypy --strict schemas/ scripts/` and `pytest tests/` pass with exit code `0`.
- `artifacts/<run_id>/FINAL_REPORT.md` is generated and verified by the Verifier for completeness and traceability back to Stage 1 IDs.

Why This Pattern Works Well in /swarm

• True Parallelism in Stages 1 & 2: Scouting across 4 user segments or analyzing 10 opportunities happens simultaneously across isolated worktrees rather than linearly in one context window.

• Zero Drift Across Stages: Because validate_stage.py checks Pydantic schemas, math invariants, and ID traceability (impl.recommendation_id -> rec.opportunity_id -> opp.id), the Verifier has objective, binary criteria to reject weak outputs before they cascade into Stage 4.

• Repeatable Replayability: Using Antigravity's /fork command on a completed 02_analyses.json, you can test completely different solution constraints (e.g., "software-only fix" vs. "hardware-plus-software fix") in Stages 3–4 without re-running discovery.

Update the Stage 2 and Stage 3 Pydantic schemas and GEMINI.md Verifier rules to require First-Principles decomposition, technical/physical contradiction framing, and Ideal Final Result (IFR) alignment.

Below are the updated schemas/pipeline.py (incorporating strict Pydantic v2 models for First-Principles decomposition, Cause-Effect Chain root causes, Technical/Physical Contradictions, and Ideal Final Result framing across Stages 2 and 3) followed by the updated GEMINI.md Adversarial Verifier rules to enforce them deterministically.

1. schemas/pipeline.py (Strict Pydantic v2 Contracts for Stages 1–4)

All models enforce ConfigDict(extra="forbid", strict=True, frozen=True) alongside custom @model_validator checks to catch hand-wavy contradictions, missing baseline assumptions, or invalid ICE math at the schema boundary.

from __future__ import annotations

from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator


class StrictBaseModel(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        strict=True,
        frozen=True,
    )


# ==============================================================================
# STAGE 1: Opportunity Discovery
# ==============================================================================

class Opportunity(StrictBaseModel):
    opportunity_id: str = Field(..., pattern=r"^OPP-\d{3}$")
    title: str = Field(..., min_length=8, max_length=140)
    industry_segment: str = Field(..., min_length=3)
    target_user_group: str = Field(..., min_length=3)
    operational_bottleneck: str = Field(..., min_length=25)
    baseline_friction_metric: str = Field(
        ...,
        min_length=15,
        description="Quantified current-state pain (e.g., '42 min/shift manual reconciliation, 8% error rate').",
    )
    supporting_evidence: list[str] = Field(..., min_length=1)


class Stage1Artifact(StrictBaseModel):
    run_id: str = Field(..., min_length=4)
    domain: str = Field(..., min_length=3)
    opportunities: list[Opportunity] = Field(..., min_length=1)

    @model_validator(mode="after")
    def verify_unique_ids(self) -> Stage1Artifact:
        ids = [o.opportunity_id for o in self.opportunities]
        if len(ids) != len(set(ids)):
            raise ValueError("Duplicate opportunity_id values detected in Stage 1.")
        return self


# ==============================================================================
# STAGE 2: First-Principles, CECA & Contradiction Analysis
# ==============================================================================

class FirstPrinciplesDecomposition(StrictBaseModel):
    fundamental_truths: list[str] = Field(
        ...,
        min_length=2,
        description="Irreducible physical, economic, or logical constraints governing the bottleneck.",
    )
    conventional_assumptions_challenged: list[str] = Field(
        ...,
        min_length=1,
        description="Inherited industry norms or workarounds falsely treated as hard constraints.",
    )
    cause_effect_chain: list[str] = Field(
        ...,
        min_length=3,
        description="Ordered causal steps from observable symptom down to the actionable root cause.",
    )
    root_cause_statement: str = Field(..., min_length=20)


class TechnicalContradiction(StrictBaseModel):
    improving_parameter: str = Field(
        ...,
        min_length=5,
        description="The system attribute that improves when a conventional fix is applied.",
    )
    worsening_parameter: str = Field(
        ...,
        min_length=5,
        description="The secondary attribute that degrades as a direct consequence.",
    )
    conflict_mechanism: str = Field(
        ...,
        min_length=20,
        description="Why improving parameter A degrades parameter B under current architecture.",
    )


class PhysicalContradiction(StrictBaseModel):
    control_element: str = Field(
        ...,
        min_length=3,
        description="The specific component, parameter, or resource experiencing opposite demands.",
    )
    state_a_requirement: str = Field(
        ...,
        min_length=10,
        description="State/value required (e.g., 'Must be rigid/high-mass to damp vibration').",
    )
    state_a_justification: str = Field(..., min_length=10)
    state_b_requirement: str = Field(
        ...,
        min_length=10,
        description="Opposite state/value required (e.g., 'Must be lightweight/compliant for portability').",
    )
    state_b_justification: str = Field(..., min_length=10)

    @model_validator(mode="after")
    def verify_distinct_opposite_states(self) -> PhysicalContradiction:
        if self.state_a_requirement.strip().lower() == self.state_b_requirement.strip().lower():
            raise ValueError("Physical contradiction requires two genuinely opposing states.")
        return self


class ICEScore(StrictBaseModel):
    impact: float = Field(..., ge=1.0, le=10.0)
    confidence: float = Field(..., ge=1.0, le=10.0)
    ease: float = Field(..., ge=1.0, le=10.0)
    composite_score: float = Field(..., ge=0.1, le=100.0)
    scoring_rationale: str = Field(..., min_length=25)

    @model_validator(mode="after")
    def verify_ice_math(self) -> ICEScore:
        expected = round((self.impact * self.confidence * self.ease) / 10.0, 2)
        if abs(round(self.composite_score, 2) - expected) > 0.05:
            raise ValueError(
                f"ICE composite_score ({self.composite_score}) != (impact*confidence*ease)/10 ({expected})."
            )
        return self


class OpportunityAnalysis(StrictBaseModel):
    analysis_id: str = Field(..., pattern=r"^ANL-\d{3}$")
    opportunity_id: str = Field(..., pattern=r"^OPP-\d{3}$")
    first_principles: FirstPrinciplesDecomposition
    technical_contradiction: TechnicalContradiction
    physical_contradiction: PhysicalContradiction
    ice_score: ICEScore
    selected_for_stage_3: bool


class Stage2Artifact(StrictBaseModel):
    run_id: str = Field(..., min_length=4)
    analyses: list[OpportunityAnalysis] = Field(..., min_length=1)

    @model_validator(mode="after")
    def verify_selection_non_empty(self) -> Stage2Artifact:
        if not any(a.selected_for_stage_3 for a in self.analyses):
            raise ValueError("At least one OpportunityAnalysis must have selected_for_stage_3=True.")
        return self


# ==============================================================================
# STAGE 3: IFR-Aligned Recommendations & Contradiction Resolution
# ==============================================================================

SeparationPrinciple = Literal[
    "separation_in_time",
    "separation_in_space",
    "separation_upon_condition",
    "separation_between_parts_and_whole",
    "system_transition_or_dynamization",
]


class IdealFinalResultSpec(StrictBaseModel):
    ifr_statement: str = Field(
        ...,
        min_length=25,
        description="Must state how the system delivers the desired function ITSELF without added cost/complexity.",
    )
    free_existing_resources_used: list[str] = Field(
        ...,
        min_length=1,
        description="Existing energy, material, information, space, or time resources harvested at near-zero marginal cost.",
    )
    eliminated_harms: list[str] = Field(..., min_length=1)
    ideality_gain_mechanism: str = Field(..., min_length=20)


class ContradictionResolution(StrictBaseModel):
    separation_strategy: SeparationPrinciple
    triz_principles_applied: list[str] = Field(
        ...,
        min_length=1,
        description="Named TRIZ inventive principles or standard solutions applied (e.g., 'Principle 15: Dynamization').",
    )
    how_contradiction_is_eliminated: str = Field(
        ...,
        min_length=25,
        description="Explains why the trade-off is resolved rather than compromised.",
    )


class TradeOffEntry(StrictBaseModel):
    risk_or_cost_factor: str = Field(..., min_length=10)
    severity: Literal["low", "medium", "high"]
    mitigation_mechanism: str = Field(..., min_length=15)


class Recommendation(StrictBaseModel):
    recommendation_id: str = Field(..., pattern=r"^REC-\d{3}$")
    opportunity_id: str = Field(..., pattern=r"^OPP-\d{3}$")
    analysis_id: str = Field(..., pattern=r"^ANL-\d{3}$")
    title: str = Field(..., min_length=8, max_length=140)
    ifr_alignment: IdealFinalResultSpec
    contradiction_resolution: ContradictionResolution
    mechanism_of_action: str = Field(..., min_length=30)
    trade_offs_and_risks: list[TradeOffEntry] = Field(..., min_length=1)
    falsifiable_validation_metric: str = Field(
        ...,
        min_length=20,
        description="Quantitative threshold proving the fix works (e.g., 'Reduce cycle time from 42m to <5m').",
    )


class Stage3Artifact(StrictBaseModel):
    run_id: str = Field(..., min_length=4)
    recommendations: list[Recommendation] = Field(..., min_length=1)


# ==============================================================================
# STAGE 4: Implementation Blueprint
# ==============================================================================

class ImplementationPhase(StrictBaseModel):
    phase_number: int = Field(..., ge=1)
    phase_name: str = Field(..., min_length=5)
    deliverables: list[str] = Field(..., min_length=1)
    verification_gate: str = Field(..., min_length=15)
    kill_switch_criteria: str = Field(..., min_length=15)


class ImplementationBlueprint(StrictBaseModel):
    blueprint_id: str = Field(..., pattern=r"^IMP-\d{3}$")
    recommendation_id: str = Field(..., pattern=r"^REC-\d{3}$")
    opportunity_id: str = Field(..., pattern=r"^OPP-\d{3}$")
    architecture_summary: str = Field(..., min_length=30)
    bill_of_materials_or_stack: list[str] = Field(..., min_length=1)
    phases: list[ImplementationPhase] = Field(..., min_length=2)

    @model_validator(mode="after")
    def verify_sequential_phases(self) -> ImplementationBlueprint:
        expected = list(range(1, len(self.phases) + 1))
        actual = [p.phase_number for p in self.phases]
        if actual != expected:
            raise ValueError(f"Phase numbers must be sequential {expected}, got {actual}.")
        return self


class Stage4Artifact(StrictBaseModel):
    run_id: str = Field(..., min_length=4)
    blueprints: list[ImplementationBlueprint] = Field(..., min_length=1)

2. Updated GEMINI.md Adversarial Verifier Rules

Replace Section 2 of your pipeline GEMINI.md with these stricter verification rules. Because the Adversarial Verifier reads GEMINI.md alongside the JSON artifacts, these rules force the Verifier to reject superficial LLM compromises (like "add an AI dashboard" or "split the difference") and demand genuine contradiction resolution.

## 2. Adversarial Verifier Rules (First-Principles, Contradiction & IFR Gates)

The Adversarial Verifier must inspect every artifact against both `schemas/pipeline.py` and the semantic rigor criteria below. Any violation is a **BLOCKING DEFECT** requiring worker revision before advancing.

### Stage 2 Gate (`02_analyses.json`)
1. **First-Principles vs. Symptom Check:**
   - Reject `first_principles.fundamental_truths` if they merely restate business wishes or software features rather than physical, thermodynamic, information-theoretic, or hard economic constraints.
   - Verify `cause_effect_chain` traces at least 3 causal hops from surface friction to an actionable root cause (`root_cause_statement`).
2. **Contradiction Integrity:**
   - **Technical Contradiction:** Confirm `improving_parameter` and `worsening_parameter` are genuinely coupled—applying a standard fix for A must demonstrably degrade B.
   - **Physical Contradiction:** Confirm `state_a_requirement` and `state_b_requirement` demand mutually exclusive states on the **exact same** `control_element` (e.g., high vs. low, continuous vs. intermittent, rigid vs. flexible). Reject vague comparisons across two different components.
3. **ICE Score Audit:** Confirm `composite_score == round((impact * confidence * ease) / 10, 2)` and verify `confidence` is penalized if Stage 1 lacked hard quantitative baseline data.

### Stage 3 Gate (`03_recommendations.json`)
1. **Ideal Final Result (IFR) Non-Bloat Check:**
   - Reject `ifr_alignment.ifr_statement` if the proposed fix solves the problem by bolting on expensive new hardware, heavy manual labor, or bloated middleware instead of making the existing system perform the function **itself**.
   - Verify `free_existing_resources_used` names concrete, already-present system resources (waste heat, idle compute, gravity, existing sensor telemetry, structural geometry, process wait states).
2. **Elimination vs. Compromise Audit:**
   - Reject recommendations that merely "balance" or "optimize" the Stage 2 contradiction (e.g., "use a medium-thickness wall"). The recommendation MUST eliminate the contradiction using the declared `separation_strategy` (time, space, condition, or part/whole) and explicit `triz_principles_applied`.
3. **Traceability & Trade-Off Honesty:**
   - Confirm `opportunity_id` and `analysis_id` match a record in `02_analyses.json` with `selected_for_stage_3 == True`.
   - Reject any recommendation whose `trade_offs_and_risks` omits realistic failure modes or lacks a concrete `mitigation_mechanism`.

### Stage 4 Gate (`04_implementations.json`)
1. **IFR Preservation:** Confirm the implementation `bill_of_materials_or_stack` does not reintroduce the bloat, cost, or complexity eliminated in Stage 3's `ifr_alignment`.
2. **Kill-Switch Falsifiability:** Confirm every phase's `kill_switch_criteria` specifies an objective quantitative abort threshold (cost, latency, error rate, or schedule slip) rather than subjective review checkpoints.

How This Changes Swarm Behavior in Practice

• Blocks "Lazy Compromise" Solutions: Without a PhysicalContradiction + SeparationPrinciple schema requirement, worker agents default to incremental middle-ground fixes. Requiring the control element to satisfy opposite states via separation in time, space, or condition forces structural inventions.

• Prevents Feature Bloat via IFR Checks: Mandating free_existing_resources_used in IdealFinalResultSpec gives the Adversarial Verifier a concrete hook to reject recommendations that add unnecessary subsystems when an existing signal, waste stream, or workflow state can do the job for free.

What is cross-stage foreign-key traceability and funnel pruning gates?

Cross-stage foreign-key traceability and funnel pruning gates are deterministic validation rules built into scripts/validate_stage.py to prevent multi-agent pipelines from hallucinating disconnected ideas, silently dropping work, or wasting tokens on low-value candidates.

Because individual Pydantic schema validation only checks a single JSON file in isolation (02_analyses.json or 03_recommendations.json), it cannot detect when an agent invents a bogus parent ID or ignores a filtering decision made in an earlier stage. These two mechanisms enforce relational integrity across files on disk.

1. Cross-Stage Foreign-Key Traceability

Just like foreign keys in a relational database (SQLite or PostgreSQL), cross-stage foreign-key traceability requires every record generated in Stage N to reference a valid, existing primary key from Stage N-1 (and back to Stage 1).

In your 4-stage pipeline, this forms a strict directed lineage chain:

Blueprint (IMP-001) --> Recommendation (REC-001) --> Analysis (ANL-003) --> Opportunity } (OPP-003)

What It Prevents in /swarm

• Phantom Parent Hallucinations: A Stage 3 worker invents a recommendation referencing OPP-099 when Stage 1 only produced OPP-001 through OPP-010.

• Mismatched Lineage Pairs: A Stage 3 recommendation pairs recommendation_id: REC-001 with opportunity_id: OPP-002 and analysis_id: ANL-005—where ANL-005 actually belonged to OPP-005.

• Orphaned Coverage / Silent Drops: Stage 2 finishes with 10 opportunities in Stage 1, but a parallel worker crashes and only 8 analyses make it into 02_analyses.json (violating 1:1 completeness between Stage 1 and Stage 2).

How validate_stage.py Enforces It

1. Stage 1 $\leftrightarrow$ Stage 2 (1:1 Completeness): Every opportunity_id in 01_opportunities.json must appear exactly once in 02_analyses.json, with zero extra or missing IDs.

2. Stage 2 $\rightarrow$ Stage 3 (1:N Validated Lineage): Every record in 03_recommendations.json must match an exact (opportunity_id, analysis_id) tuple present in 02_analyses.json.

3. Stage 3 $\rightarrow$ Stage 4 (1:1 or 1:N Full Coverage): Every recommendation_id in 03_recommendations.json must map cleanly to a corresponding ImplementationBlueprint in 04_implementations.json with matching opportunity_id.

2. Funnel Pruning Gates

A multi-stage innovation pipeline intentionally starts wide and narrows down: you might discover 12 opportunities in Stage 1, analyze all 12 in Stage 2, and advance only the top 3 into deep TRIZ contradiction resolution (Stage 3) and implementation architecture (Stage 4).

Funnel pruning gates enforce both the selection cap and the inclusion/exclusion boundary between stages.

What It Prevents in /swarm

• Scope Creep / Token Burn: Stage 2 marks 8 opportunities as selected_for_stage_3: true instead of the budgeted top 3, tripling token consumption in Stages 3 and 4.

• Gate Bypassing: A Stage 3 worker generates a recommendation for OPP-004 even though Stage 2 scored it poorly and set selected_for_stage_3: false.

• Rank Inversion (Cherry-Picking): Stage 2 marks OPP-007 (ICE score 28.0) as selected_for_stage_3: true while leaving OPP-002 (ICE score 72.0) unselected without an explicit override flag.

How validate_stage.py Enforces It

Gate Check	Rule Enforced	Failure Trigger
Selection Cardinality	sum(a.selected_for_stage_3 for a in analyses) == max_selected (e.g., 3)	Too few or too many opportunities flagged to advance.
Rank Monotonicity	Every selected analysis must have composite_score >= every unselected analysis (unless explicitly flagged as a strategic override).	Coordinator selects a lower-scoring candidate over a higher-scoring one.
Pruned Exclusion	Any opportunity_id with selected_for_stage_3 == False is forbidden from appearing in 03_recommendations.json or 04_implementations.json.	Worker wastes effort generating fixes for a rejected opportunity.
Selected Completeness	Every opportunity_id with selected_for_stage_3 == True must have at least one verified Recommendation in 03_recommendations.json.	Worker drops one of the top 3 selected opportunities before Stage 3 finishes.

Why These Two Mechanisms Matter Together

Pydantic guarantees structural validity (every field has the right type and non-empty string length). Cross-stage foreign-key traceability and funnel pruning gates guarantee relational and procedural integrity—ensuring that the final FINAL_REPORT.md traces every implementation step back to a verified, top-ranked root cause without gaps, orphans, or wasted swarm quota.

Yes, write the complete scripts/validate_stage.py file implementing Pydantic schema validation, cross-stage foreign-key traceability checks, and funnel pruning gates.

Below is the complete, standalone scripts/validate_stage.py CLI gatekeeper. It validates JSON artifacts against schemas/pipeline.py, checks cross-stage foreign-key lineage (IMP -> REC -> ANL -> OPP), enforces Stage 2 funnel pruning cardinality and rank monotonicity, and prints machine-readable defect diagnostics with deterministic exit codes (0 for pass, 1 for validation failure, 2 for missing file/CLI error) so the /swarm Verifier can gate stage transitions cleanly.

scripts/validate_stage.py

#!/usr/bin/env python3
"""Deterministic Stage Gatekeeper for the 4-Stage Innovation Pipeline.

Enforces:
1. Strict Pydantic v2 schema validation per stage artifact.
2. Cross-stage foreign-key lineage integrity (IMP -> REC -> ANL -> OPP).
3. Funnel pruning gates (selection cardinality, rank monotonicity, pruned exclusion).

Exit Codes:
  0 - All schema, relational, and funnel checks passed.
  1 - One or more validation defects detected (blocking).
  2 - Missing artifact file, malformed JSON, or invalid CLI invocation.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import TypeVar

from pydantic import BaseModel, ValidationError

# Ensure repository root is importable when executed as `python scripts/validate_stage.py`
REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from schemas.pipeline import (  # noqa: E402
    Stage1Artifact,
    Stage2Artifact,
    Stage3Artifact,
    Stage4Artifact,
)

TModel = TypeVar("TModel", bound=BaseModel)


class GateFailure(Exception):
    """Raised when a cross-stage relational or funnel invariant is violated."""


def load_and_validate_model(file_path: Path, model_cls: type[TModel]) -> TModel:
    """Load JSON from disk and validate strictly against the target Pydantic v2 model."""
    if not file_path.is_file():
        raise FileNotFoundError(f"Required artifact missing: {file_path}")

    try:
        raw_text = file_path.read_text(encoding="utf-8")
        payload = json.loads(raw_text)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Malformed JSON in {file_path}: {exc}") from exc

    return model_cls.model_validate(payload)


def validate_stage_1(artifacts_dir: Path, run_id: str) -> Stage1Artifact:
    s1 = load_and_validate_model(artifacts_dir / "01_opportunities.json", Stage1Artifact)
    if s1.run_id != run_id:
        raise GateFailure(
            f"[Stage 1] Artifact run_id '{s1.run_id}' does not match expected '{run_id}'."
        )
    return s1


def validate_stage_2(
    artifacts_dir: Path,
    run_id: str,
    max_selected: int,
    enforce_rank_order: bool,
) -> tuple[Stage1Artifact, Stage2Artifact]:
    s1 = validate_stage_1(artifacts_dir, run_id)
    s2 = load_and_validate_model(artifacts_dir / "02_analyses.json", Stage2Artifact)

    if s2.run_id != run_id:
        raise GateFailure(
            f"[Stage 2] Artifact run_id '{s2.run_id}' does not match expected '{run_id}'."
        )

    errors: list[str] = []

    # 1. Unique analysis_id check
    anl_ids = [a.analysis_id for a in s2.analyses]
    if len(anl_ids) != len(set(anl_ids)):
        errors.append("[Stage 2] Duplicate analysis_id values detected.")

    # 2. Cross-stage 1:1 completeness check (Stage 1 OPP <-> Stage 2 ANL)
    s1_opp_ids = {o.opportunity_id for o in s1.opportunities}
    s2_opp_ids = [a.opportunity_id for a in s2.analyses]

    if len(s2_opp_ids) != len(set(s2_opp_ids)):
        errors.append("[Stage 2] Multiple analyses reference the same opportunity_id.")

    missing_opps = sorted(s1_opp_ids - set(s2_opp_ids))
    phantom_opps = sorted(set(s2_opp_ids) - s1_opp_ids)

    if missing_opps:
        errors.append(
            f"[Stage 2 FK Violation] Missing analyses for Stage 1 opportunities: {missing_opps}"
        )
    if phantom_opps:
        errors.append(
            f"[Stage 2 FK Violation] Analyses reference non-existent Stage 1 opportunities: {phantom_opps}"
        )

    # 3. Funnel pruning cardinality check
    selected = [a for a in s2.analyses if a.selected_for_stage_3]
    unselected = [a for a in s2.analyses if not a.selected_for_stage_3]

    if len(selected) > max_selected:
        errors.append(
            f"[Stage 2 Funnel Violation] Selected {len(selected)} opportunities, exceeding cap of {max_selected}."
        )
    if len(selected) == 0:
        errors.append("[Stage 2 Funnel Violation] Zero opportunities selected for Stage 3.")

    # 4. Funnel rank monotonicity check (no lower-scoring candidate selected over a higher-scoring one)
    if enforce_rank_order and selected and unselected:
        min_selected_score = min(a.ice_score.composite_score for a in selected)
        max_unselected_score = max(a.ice_score.composite_score for a in unselected)
        if min_selected_score < max_unselected_score:
            bad_selected = [
                f"{a.opportunity_id}({a.ice_score.composite_score})"
                for a in selected
                if a.ice_score.composite_score < max_unselected_score
            ]
            higher_unselected = [
                f"{a.opportunity_id}({a.ice_score.composite_score})"
                for a in unselected
                if a.ice_score.composite_score > min_selected_score
            ]
            errors.append(
                "[Stage 2 Rank Inversion] Lower-scoring opportunities were selected ahead of "
                f"higher-scoring candidates. Selected: {bad_selected}; Unselected higher: {higher_unselected}."
            )

    if errors:
        raise GateFailure("\n".join(errors))

    return s1, s2


def validate_stage_3(
    artifacts_dir: Path,
    run_id: str,
    max_selected: int,
    enforce_rank_order: bool,
) -> tuple[Stage1Artifact, Stage2Artifact, Stage3Artifact]:
    s1, s2 = validate_stage_2(artifacts_dir, run_id, max_selected, enforce_rank_order)
    s3 = load_and_validate_model(artifacts_dir / "03_recommendations.json", Stage3Artifact)

    if s3.run_id != run_id:
        raise GateFailure(
            f"[Stage 3] Artifact run_id '{s3.run_id}' does not match expected '{run_id}'."
        )

    errors: list[str] = []

    # 1. Unique recommendation_id check
    rec_ids = [r.recommendation_id for r in s3.recommendations]
    if len(rec_ids) != len(set(rec_ids)):
        errors.append("[Stage 3] Duplicate recommendation_id values detected.")

    # Lookup maps from Stage 2
    anl_by_id = {a.analysis_id: a for a in s2.analyses}
    selected_opp_ids = {a.opportunity_id for a in s2.analyses if a.selected_for_stage_3}
    pruned_opp_ids = {a.opportunity_id for a in s2.analyses if not a.selected_for_stage_3}

    covered_selected_opps: set[str] = set()

    # 2. Foreign-key lineage & pruned-exclusion checks
    for rec in s3.recommendations:
        anl = anl_by_id.get(rec.analysis_id)
        if anl is None:
            errors.append(
                f"[Stage 3 FK Violation] {rec.recommendation_id} references unknown analysis_id '{rec.analysis_id}'."
            )
            continue

        if anl.opportunity_id != rec.opportunity_id:
            errors.append(
                f"[Stage 3 Lineage Mismatch] {rec.recommendation_id} pairs opportunity_id "
                f"'{rec.opportunity_id}' with '{rec.analysis_id}' (which belongs to '{anl.opportunity_id}')."
            )

        if rec.opportunity_id in pruned_opp_ids:
            errors.append(
                f"[Stage 3 Funnel Leak] {rec.recommendation_id} targets pruned opportunity "
                f"'{rec.opportunity_id}' (selected_for_stage_3=False)."
            )
        elif rec.opportunity_id in selected_opp_ids:
            covered_selected_opps.add(rec.opportunity_id)

    # 3. Ensure every Stage 2 selected opportunity received at least one recommendation
    uncovered_selected = sorted(selected_opp_ids - covered_selected_opps)
    if uncovered_selected:
        errors.append(
            "[Stage 3 Coverage Violation] Stage 2 selected opportunities missing recommendations: "
            f"{uncovered_selected}"
        )

    if errors:
        raise GateFailure("\n".join(errors))

    return s1, s2, s3


def validate_stage_4(
    artifacts_dir: Path,
    run_id: str,
    max_selected: int,
    enforce_rank_order: bool,
) -> tuple[Stage1Artifact, Stage2Artifact, Stage3Artifact, Stage4Artifact]:
    s1, s2, s3 = validate_stage_3(artifacts_dir, run_id, max_selected, enforce_rank_order)
    s4 = load_and_validate_model(artifacts_dir / "04_implementations.json", Stage4Artifact)

    if s4.run_id != run_id:
        raise GateFailure(
            f"[Stage 4] Artifact run_id '{s4.run_id}' does not match expected '{run_id}'."
        )

    errors: list[str] = []

    # 1. Unique blueprint_id check
    imp_ids = [b.blueprint_id for b in s4.blueprints]
    if len(imp_ids) != len(set(imp_ids)):
        errors.append("[Stage 4] Duplicate blueprint_id values detected.")

    rec_by_id = {r.recommendation_id: r for r in s3.recommendations}
    covered_rec_ids: set[str] = set()

    # 2. Foreign-key lineage check (IMP -> REC -> OPP)
    for bp in s4.blueprints:
        rec = rec_by_id.get(bp.recommendation_id)
        if rec is None:
            errors.append(
                f"[Stage 4 FK Violation] {bp.blueprint_id} references unknown recommendation_id '{bp.recommendation_id}'."
            )
            continue

        if rec.opportunity_id != bp.opportunity_id:
            errors.append(
                f"[Stage 4 Lineage Mismatch] {bp.blueprint_id} pairs opportunity_id "
                f"'{bp.opportunity_id}' with '{bp.recommendation_id}' (which belongs to '{rec.opportunity_id}')."
            )

        covered_rec_ids.add(bp.recommendation_id)

    # 3. Full coverage check across all Stage 3 recommendations
    missing_recs = sorted(set(rec_by_id.keys()) - covered_rec_ids)
    if missing_recs:
        errors.append(
            "[Stage 4 Coverage Violation] Stage 3 recommendations missing implementation blueprints: "
            f"{missing_recs}"
        )

    if errors:
        raise GateFailure("\n".join(errors))

    return s1, s2, s3, s4


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Validate stage schemas, cross-stage foreign keys, and funnel pruning gates."
    )
    parser.add_argument(
        "--run-id",
        required=True,
        help="Run identifier matching artifacts/<run-id>/ directory and artifact run_id field.",
    )
    parser.add_argument(
        "--stage",
        type=int,
        choices=[1, 2, 3, 4],
        required=True,
        help="Target stage to validate (validates all prerequisite stages transitively).",
    )
    parser.add_argument(
        "--artifacts-root",
        type=Path,
        default=REPO_ROOT / "artifacts",
        help="Root directory containing run artifact folders (default: ./artifacts).",
    )
    parser.add_argument(
        "--max-selected",
        type=int,
        default=3,
        help="Maximum opportunities allowed to pass the Stage 2 funnel pruning gate (default: 3).",
    )
    parser.add_argument(
        "--allow-rank-override",
        action="store_true",
        help="Disable strict ICE rank monotonicity check during Stage 2 funnel selection.",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    run_dir: Path = args.artifacts_root / args.run_id
    enforce_rank = not args.allow_rank_override

    try:
        if args.stage == 1:
            s1 = validate_stage_1(run_dir, args.run_id)
            print(
                f"[PASS] Stage 1 verified for '{args.run_id}': "
                f"{len(s1.opportunities)} opportunities validated."
            )
        elif args.stage == 2:
            s1, s2 = validate_stage_2(run_dir, args.run_id, args.max_selected, enforce_rank)
            sel_count = sum(1 for a in s2.analyses if a.selected_for_stage_3)
            print(
                f"[PASS] Stage 2 verified for '{args.run_id}': "
                f"{len(s2.analyses)}/{len(s1.opportunities)} analyzed, "
                f"{sel_count} selected for Stage 3."
            )
        elif args.stage == 3:
            _, s2, s3 = validate_stage_3(run_dir, args.run_id, args.max_selected, enforce_rank)
            sel_count = sum(1 for a in s2.analyses if a.selected_for_stage_3)
            print(
                f"[PASS] Stage 3 verified for '{args.run_id}': "
                f"{len(s3.recommendations)} recommendations across {sel_count} selected opportunities."
            )
        elif args.stage == 4:
            _, _, s3, s4 = validate_stage_4(run_dir, args.run_id, args.max_selected, enforce_rank)
            print(
                f"[PASS] Stage 4 verified for '{args.run_id}': "
                f"{len(s4.blueprints)} implementation blueprints covering {len(s3.recommendations)} recommendations."
            )
        return 0

    except (FileNotFoundError, ValueError) as exc:
        print(f"[ERROR] Input/JSON failure:\n{exc}", file=sys.stderr)
        return 2
    except ValidationError as exc:
        print(f"[FAIL] Pydantic schema validation error:\n{exc}", file=sys.stderr)
        return 1
    except GateFailure as exc:
        print(f"[FAIL] Cross-stage relational / funnel gate violation:\n{exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

Key Design Details for /swarm Execution

• Transitive Validation: Running --stage 4 automatically re-verifies Stages 1, 2, and 3 first. If a late-stage worker accidentally mutates an upstream JSON file while resolving a merge conflict, the gate immediately catches the regression.

• Actionable stderr Output: Every error message prints the exact offending IDs (OPP-xxx, ANL-xxx, REC-xxx, IMP-xxx) and mismatch type (FK Violation, Lineage Mismatch, Funnel Leak, Rank Inversion), allowing the /swarm Coordinator to route targeted repair instructions to the exact worker responsible.

• Configurable Funnel Cap: Passing --max-selected 5 or --allow-rank-override directly from your /swarm prompt lets you adjust funnel width per run without touching Python source files.

Yes, write scripts/render_report.py to validate Stage 4 artifacts and render the complete, lineage-traced FINAL_REPORT.md dossier.

Below is the complete scripts/render_report.py script. It calls validate_stage_4() directly from scripts/validate_stage.py before rendering a single line of prose—guaranteeing that FINAL_REPORT.md can never be generated from unverified, orphaned, or schema-broken artifacts.

scripts/render_report.py

#!/usr/bin/env python3
"""Deterministic Markdown Report Compiler for the 4-Stage Innovation Pipeline.

Runs full transitive validation via `validate_stage_4` before joining Stage 1–4
Pydantic models by foreign key and rendering `artifacts/<run-id>/FINAL_REPORT.md`.

Exit Codes:
  0 - Validation passed and FINAL_REPORT.md written successfully.
  1 - Validation gate failure (blocking).
  2 - Missing artifact file, malformed JSON, or invalid CLI invocation.
"""

from __future__ import annotations

import argparse
import sys
from datetime import datetime, timezone
from pathlib import Path

from pydantic import ValidationError

REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from schemas.pipeline import (  # noqa: E402
    ImplementationBlueprint,
    Opportunity,
    OpportunityAnalysis,
    Recommendation,
    Stage1Artifact,
    Stage2Artifact,
    Stage3Artifact,
    Stage4Artifact,
)
from scripts.validate_stage import GateFailure, validate_stage_4  # noqa: E402


def render_bullet_list(items: list[str], indent: str = "") -> str:
    return "\n".join(f"{indent}- {item}" for item in items)


def render_numbered_list(items: list[str], indent: str = "") -> str:
    return "\n".join(f"{indent}{idx}. {item}" for idx, item in enumerate(items, start=1))


def build_markdown_dossier(
    s1: Stage1Artifact,
    s2: Stage2Artifact,
    s3: Stage3Artifact,
    s4: Stage4Artifact,
) -> str:
    """Join validated Stage 1-4 models by foreign key and render a traceable Markdown dossier."""
    timestamp_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")

    opp_by_id: dict[str, Opportunity] = {o.opportunity_id: o for o in s1.opportunities}
    anl_by_opp_id: dict[str, OpportunityAnalysis] = {a.opportunity_id: a for a in s2.analyses}

    recs_by_opp_id: dict[str, list[Recommendation]] = {}
    for rec in s3.recommendations:
        recs_by_opp_id.setdefault(rec.opportunity_id, []).append(rec)

    bps_by_rec_id: dict[str, list[ImplementationBlueprint]] = {}
    for bp in s4.blueprints:
        bps_by_rec_id.setdefault(bp.recommendation_id, []).append(bp)

    sorted_analyses = sorted(
        s2.analyses,
        key=lambda a: (not a.selected_for_stage_3, -a.ice_score.composite_score, a.opportunity_id),
    )
    selected_analyses = [a for a in sorted_analyses if a.selected_for_stage_3]
    pruned_analyses = [a for a in sorted_analyses if not a.selected_for_stage_3]

    lines: list[str] = []

    # --------------------------------------------------------------------------
    # Header & Executive Funnel Summary
    # --------------------------------------------------------------------------
    lines.extend(
        [
            f"# Innovation Discovery & Implementation Dossier — `{s1.run_id}`",
            "",
            f"- **Domain:** {s1.domain}",
            f"- **Generated:** {timestamp_utc}",
            f"- **Funnel Summary:** {len(s1.opportunities)} Discovered → "
            f"{len(s2.analyses)} Analyzed → "
            f"{len(selected_analyses)} Selected → "
            f"{len(s3.recommendations)} Recommendations → "
            f"{len(s4.blueprints)} Implementation Blueprints",
            "",
            "---",
            "",
            "## 1. Executive Triage & Funnel Matrix",
            "",
            "| Status | Opp ID | Analysis ID | Title | Target User Group | ICE Score (I / C / E) |",
            "| :--- | :--- | :--- | :--- | :--- | :--- |",
        ]
    )

    for anl in sorted_analyses:
        opp = opp_by_id[anl.opportunity_id]
        status = "**SELECTED**" if anl.selected_for_stage_3 else "Pruned"
        ice = anl.ice_score
        score_str = f"**{ice.composite_score:.2f}** ({ice.impact:.1f} / {ice.confidence:.1f} / {ice.ease:.1f})"
        lines.append(
            f"| {status} | `{opp.opportunity_id}` | `{anl.analysis_id}` | "
            f"{opp.title} | {opp.target_user_group} | {score_str} |"
        )

    lines.extend(["", "---", "", "## 2. Selected Opportunities — Deep-Dive Lineage", ""])

    # --------------------------------------------------------------------------
    # Full Lineage Dossiers for Selected Candidates (OPP -> ANL -> REC -> IMP)
    # --------------------------------------------------------------------------
    for anl in selected_analyses:
        opp = opp_by_id[anl.opportunity_id]
        fp = anl.first_principles
        tc = anl.technical_contradiction
        pc = anl.physical_contradiction
        ice = anl.ice_score
        recs = recs_by_opp_id.get(opp.opportunity_id, [])

        lines.extend(
            [
                f"### [{opp.opportunity_id}] {opp.title}",
                "",
                f"> **Lineage:** `{opp.opportunity_id}` → `{anl.analysis_id}` → "
                + ", ".join(f"`{r.recommendation_id}`" for r in recs),
                "",
                "#### Stage 1: Operational Bottleneck & Baseline",
                f"- **Industry Segment:** {opp.industry_segment}",
                f"- **Target User Group:** {opp.target_user_group}",
                f"- **Operational Bottleneck:** {opp.operational_bottleneck}",
                f"- **Baseline Friction Metric:** {opp.baseline_friction_metric}",
                "- **Supporting Evidence:**",
                render_bullet_list(opp.supporting_evidence, indent="  "),
                "",
                f"#### Stage 2: First-Principles & Contradiction Analysis (`{anl.analysis_id}`)",
                f"- **ICE Composite Score:** **{ice.composite_score:.2f}** "
                f"(Impact: `{ice.impact}`, Confidence: `{ice.confidence}`, Ease: `{ice.ease}`)",
                f"- **Scoring Rationale:** {ice.scoring_rationale}",
                "- **Fundamental Truths (Irreducible Constraints):**",
                render_bullet_list(fp.fundamental_truths, indent="  "),
                "- **Conventional Assumptions Challenged:**",
                render_bullet_list(fp.conventional_assumptions_challenged, indent="  "),
                "- **Cause-Effect Chain (Symptom → Root Cause):**",
                render_numbered_list(fp.cause_effect_chain, indent="  "),
                f"- **Actionable Root Cause:** {fp.root_cause_statement}",
                "",
                "**Contradiction Framing:**",
                "| Contradiction Type | Formulation |",
                "| :--- | :--- |",
                f"| **Technical Contradiction** | Improving **{tc.improving_parameter}** degrades **{tc.worsening_parameter}**. *Mechanism:* {tc.conflict_mechanism} |",
                f"| **Physical Contradiction** | Control element **{pc.control_element}** must be **{pc.state_a_requirement}** ({pc.state_a_justification}) AND **{pc.state_b_requirement}** ({pc.state_b_justification}). |",
                "",
            ]
        )

        for rec in recs:
            ifr = rec.ifr_alignment
            cr = rec.contradiction_resolution
            blueprints = bps_by_rec_id.get(rec.recommendation_id, [])

            lines.extend(
                [
                    f"#### Stage 3: IFR-Aligned Recommendation (`{rec.recommendation_id}` → `{anl.analysis_id}`)",
                    f"**{rec.title}**",
                    "",
                    f"- **Ideal Final Result (IFR):** {ifr.ifr_statement}",
                    f"- **Ideality Gain Mechanism:** {ifr.ideality_gain_mechanism}",
                    "- **Free Existing Resources Harvested:**",
                    render_bullet_list(ifr.free_existing_resources_used, indent="  "),
                    "- **Eliminated Harms:**",
                    render_bullet_list(ifr.eliminated_harms, indent="  "),
                    f"- **Separation Strategy:** `{cr.separation_strategy}`",
                    f"- **TRIZ Principles Applied:** {', '.join(cr.triz_principles_applied)}",
                    f"- **Contradiction Elimination:** {cr.how_contradiction_is_eliminated}",
                    f"- **Mechanism of Action:** {rec.mechanism_of_action}",
                    f"- **Falsifiable Validation Metric:** `{rec.falsifiable_validation_metric}`",
                    "",
                    "**Trade-Offs & Risk Mitigations:**",
                    "| Risk / Cost Factor | Severity | Mitigation Mechanism |",
                    "| :--- | :--- | :--- |",
                ]
            )

            for tr in rec.trade_offs_and_risks:
                lines.append(
                    f"| {tr.risk_or_cost_factor} | `{tr.severity.upper()}` | {tr.mitigation_mechanism} |"
                )

            lines.append("")

            for bp in blueprints:
                lines.extend(
                    [
                        f"#### Stage 4: Implementation Blueprint (`{bp.blueprint_id}` → `{rec.recommendation_id}`)",
                        f"- **Architecture Summary:** {bp.architecture_summary}",
                        "- **Bill of Materials / Technical Stack:**",
                        render_bullet_list(bp.bill_of_materials_or_stack, indent="  "),
                        "",
                        "| Phase | Name | Deliverables | Verification Gate | Kill-Switch Criteria |",
                        "| :--- | :--- | :--- | :--- | :--- |",
                    ]
                )

                for ph in bp.phases:
                    deliv_str = "; ".join(ph.deliverables)
                    lines.append(
                        f"| Phase {ph.phase_number} | **{ph.phase_name}** | {deliv_str} | "
                        f"{ph.verification_gate} | {ph.kill_switch_criteria} |"
                    )

                lines.append("")

        lines.extend(["---", ""])

    # --------------------------------------------------------------------------
    # Appendix: Pruned Opportunities Audit Log
    # --------------------------------------------------------------------------
    if pruned_analyses:
        lines.extend(
            [
                "## 3. Appendix — Pruned Opportunities Audit Log",
                "",
                "The following opportunities were analyzed in Stage 2 and excluded from Stages 3–4 by the funnel pruning gate:",
                "",
            ]
        )
        for anl in pruned_analyses:
            opp = opp_by_id[anl.opportunity_id]
            lines.extend(
                [
                    f"- **`{opp.opportunity_id}` / `{anl.analysis_id}` — {opp.title}** "
                    f"(ICE: `{anl.ice_score.composite_score:.2f}`): "
                    f"*Root Cause:* {anl.first_principles.root_cause_statement} "
                    f"*Rationale:* {anl.ice_score.scoring_rationale}",
                ]
            )
        lines.append("")

    return "\n".join(lines)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Validate Stage 4 artifacts and compile FINAL_REPORT.md."
    )
    parser.add_argument(
        "--run-id",
        required=True,
        help="Run identifier matching artifacts/<run-id>/ directory.",
    )
    parser.add_argument(
        "--artifacts-root",
        type=Path,
        default=REPO_ROOT / "artifacts",
        help="Root directory containing run artifact folders (default: ./artifacts).",
    )
    parser.add_argument(
        "--max-selected",
        type=int,
        default=3,
        help="Maximum opportunities allowed to pass the Stage 2 funnel pruning gate (default: 3).",
    )
    parser.add_argument(
        "--allow-rank-override",
        action="store_true",
        help="Disable strict ICE rank monotonicity check during Stage 2 funnel selection.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=None,
        help="Optional custom output path (default: artifacts/<run-id>/FINAL_REPORT.md).",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    run_dir: Path = args.artifacts_root / args.run_id
    output_path: Path = args.output or (run_dir / "FINAL_REPORT.md")
    enforce_rank = not args.allow_rank_override

    try:
        s1, s2, s3, s4 = validate_stage_4(
            artifacts_dir=run_dir,
            run_id=args.run_id,
            max_selected=args.max_selected,
            enforce_rank_order=enforce_rank,
        )
    except (FileNotFoundError, ValueError) as exc:
        print(f"[ERROR] Input/JSON failure:\n{exc}", file=sys.stderr)
        return 2
    except ValidationError as exc:
        print(f"[FAIL] Pydantic schema validation error:\n{exc}", file=sys.stderr)
        return 1
    except GateFailure as exc:
        print(f"[FAIL] Cross-stage relational / funnel gate violation:\n{exc}", file=sys.stderr)
        return 1

    markdown_output = build_markdown_dossier(s1, s2, s3, s4)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(markdown_output, encoding="utf-8")

    print(
        f"[PASS] Verified all 4 stages for '{args.run_id}' and compiled report to:\n"
        f"       {output_path}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

How This Integrates with /swarm

• Single-Command Gate + Render: In Stage 4 of your /swarm prompt, the worker runs python scripts/render_report.py --run-id <run_id>. Because render_report.py calls validate_stage_4() internally, a single command verifies all four JSON files, checks every foreign key, confirms funnel pruning compliance, and writes FINAL_REPORT.md.

• Complete Auditability: Pruned Stage 2 opportunities are preserved in Section 3 (Appendix) alongside their ICE scores and root-cause statements so you can inspect why candidates were dropped without digging through raw JSON files.

Write scripts/run_local_ollama.py using Pydantic v2 structured outputs and local Ollama models to execute the exact same 4-stage pipeline and validate_stage.py gates outside Antigravity.

Below is the complete, standalone scripts/run_local_ollama.py orchestrator. It mirrors the exact role-to-model routing, adversarial critic loops, Pydantic v2 structured output enforcement (format=Model.model_json_schema()), and deterministic validate_stage.py / render_report.py gates of your /swarm setup—allowing you to run the identical 4-stage innovation pipeline locally against your Ollama server (http://localhost:11434).

scripts/run_local_ollama.py

#!/usr/bin/env python3
"""Local Multi-Stage Innovation Pipeline Runner powered by Ollama & Pydantic v2.

Executes the exact 4-stage First-Principles / TRIZ / IFR innovation pipeline
locally using Ollama structured JSON outputs, adversarial critic verification
loops, deterministic `validate_stage.py` gates, and `render_report.py`.

Default Model Routing (mirrors /swarm heterogeneous roles):
  - Worker / Scout / Analyst: granite4:latest (or any local Ollama model)
  - Adversarial Critic / Blueprint Architect: nemotron-3:30b (or deep-reasoning model)

Usage:
  python scripts/run_local_ollama.py \
    --run-id 2026-11-13_marina_ops \
    --domain "Coastal Marina & Dock Logistics" \
    --user-groups "Small marina operators, vessel owners, marine technicians" \
    --constraints "CapEx under $5,000 per unit; zero cloud dependency dockside"
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Literal, TypeVar

from pydantic import BaseModel, ConfigDict, Field, ValidationError

REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from schemas.pipeline import (  # noqa: E402
    ImplementationBlueprint,
    Opportunity,
    OpportunityAnalysis,
    Recommendation,
    Stage1Artifact,
    Stage2Artifact,
    Stage3Artifact,
    Stage4Artifact,
)
from scripts.render_report import build_markdown_dossier  # noqa: E402
from scripts.validate_stage import (  # noqa: E402
    GateFailure,
    validate_stage_1,
    validate_stage_2,
    validate_stage_3,
    validate_stage_4,
)

TModel = TypeVar("TModel", bound=BaseModel)


# ==============================================================================
# Helper Models for Per-Item Parallel-Style Generation & Adversarial Critique
# ==============================================================================

class OpportunityDraftBatch(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    opportunities: list[Opportunity] = Field(..., min_length=1)


class CriticVerdict(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    status: Literal["PASS", "REJECT"]
    blocking_defects: list[str] = Field(
        default_factory=list,
        description="Concrete First-Principles, Contradiction, IFR, or schema defects if REJECT.",
    )
    revision_instructions: str = Field(
        default="",
        description="Actionable fix instructions for the worker if status == REJECT.",
    )


# ==============================================================================
# Deterministic Ollama HTTP Client with Pydantic Structured Output & Repair Loop
# ==============================================================================

class OllamaStructuredClient:
    """Zero-dependency HTTP client for Ollama /api/chat with JSON schema enforcement."""

    def __init__(self, base_url: str = "http://localhost:11434", timeout_sec: int = 300) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout_sec = timeout_sec

    def _post_chat(
        self,
        model: str,
        messages: list[dict[str, str]],
        json_schema: dict[str, Any],
        temperature: float,
    ) -> str:
        endpoint = f"{self.base_url}/api/chat"
        payload = {
            "model": model,
            "messages": messages,
            "stream": False,
            "format": json_schema,
            "options": {
                "temperature": temperature,
                "num_ctx": 16384,
            },
        }
        req = urllib.request.Request(
            endpoint,
            data=json.dumps(payload).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(req, timeout=self.timeout_sec) as resp:
                body = json.loads(resp.read().decode("utf-8"))
        except urllib.error.URLError as exc:
            raise RuntimeError(
                f"Failed to connect to Ollama at {endpoint}. Ensure `ollama serve` is running. Error: {exc}"
            ) from exc

        message = body.get("message", {})
        content = message.get("content", "")
        if not content:
            raise RuntimeError(f"Empty response returned from Ollama model '{model}'.")
        return str(content)

    def generate_structured(
        self,
        *,
        model: str,
        system_prompt: str,
        user_prompt: str,
        schema_cls: type[TModel],
        max_retries: int = 3,
        temperature: float = 0.2,
    ) -> TModel:
        """Generate and validate a Pydantic v2 model with automatic schema-error repair."""
        schema_dict = schema_cls.model_json_schema()
        messages: list[dict[str, str]] = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ]

        last_error = ""
        for attempt in range(1, max_retries + 1):
            raw_json_str = self._post_chat(
                model=model,
                messages=messages,
                json_schema=schema_dict,
                temperature=temperature,
            )
            try:
                parsed = json.loads(raw_json_str)
                return schema_cls.model_validate(parsed)
            except (json.JSONDecodeError, ValidationError) as exc:
                last_error = str(exc)
                print(
                    f"  [Schema Repair {attempt}/{max_retries}] {schema_cls.__name__} failed validation; feeding error back...",
                    file=sys.stderr,
                )
                messages.append({"role": "assistant", "content": raw_json_str})
                messages.append(
                    {
                        "role": "user",
                        "content": (
                            "Your JSON output failed strict Pydantic v2 validation. "
                            "Fix EVERY error listed below and return valid JSON matching the schema:\n\n"
                            f"{last_error}"
                        ),
                    }
                )

        raise RuntimeError(
            f"Exhausted {max_retries} schema repair attempts for {schema_cls.__name__}:\n{last_error}"
        )


# ==============================================================================
# System Prompts Enforcing GEMINI.md Analytical & Adversarial Standards
# ==============================================================================

WORKER_SYSTEM_PROMPT = """You are a First-Principles Systems Architect and TRIZ Innovation Engineer.
Rules:
1. Never use vague marketing fluff ('leverage AI', 'streamline workflows'). Cite concrete physical, software, or operational mechanisms.
2. Ensure all IDs strictly match required regex patterns (OPP-001, ANL-001, REC-001, IMP-001).
3. In Stage 2, compute ICE composite_score strictly as: round((impact * confidence * ease) / 10.0, 2).
4. In Stage 2, Physical Contradiction must specify two genuinely opposing states on the EXACT SAME control_element.
5. In Stage 3, Ideal Final Result (IFR) must explain how the existing system achieves the function ITSELF using free existing resources rather than bolting on expensive bloat.
6. Return ONLY valid JSON adhering strictly to the provided JSON Schema."""

CRITIC_SYSTEM_PROMPT = """You are an Adversarial Verification Auditor enforcing strict engineering rigor.
Audit the candidate artifact against these blocking gates:
1. First-Principles Check: Are fundamental_truths real physical/economic constraints rather than feature wishes? Does cause_effect_chain trace >= 3 causal hops?
2. Contradiction Check: Does Technical Contradiction couple an improving parameter to a worsening parameter? Does Physical Contradiction demand opposite states on the SAME control_element?
3. IFR & Anti-Bloat Check: Does the recommendation solve the bottleneck using free existing resources and TRIZ separation principles rather than adding expensive hardware/software bloat?
4. Falsifiability Check: Are baseline metrics, validation metrics, and phase kill-switch criteria concrete and quantitative?

If ANY defect exists, set status='REJECT' and list exact defects and revision instructions. Only set status='PASS' when all gates are satisfied."""


# ==============================================================================
# 4-Stage Pipeline Execution
# ==============================================================================

def write_json(path: Path, model: BaseModel) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(model.model_dump_json(indent=2) + "\n", encoding="utf-8")


def run_stage_1(
    client: OllamaStructuredClient,
    worker_model: str,
    run_dir: Path,
    run_id: str,
    domain: str,
    user_groups: list[str],
    constraints: str,
    opps_per_group: int,
) -> Stage1Artifact:
    print(f"\n=== STAGE 1: Opportunity Discovery ({len(user_groups)} user groups) ===")
    all_opps: list[Opportunity] = []
    counter = 1

    for group in user_groups:
        print(f"  -> Scouting user group: '{group}'...")
        prompt = (
            f"Target Domain: {domain}\n"
            f"Target User Group: {group}\n"
            f"Hard Constraints: {constraints}\n\n"
            f"Identify {opps_per_group} high-friction operational bottlenecks faced by '{group}'. "
            f"Assign temporary IDs starting at OPP-{counter:03d}. "
            "Every opportunity must include a quantified baseline_friction_metric and concrete supporting_evidence."
        )
        batch = client.generate_structured(
            model=worker_model,
            system_prompt=WORKER_SYSTEM_PROMPT,
            user_prompt=prompt,
            schema_cls=OpportunityDraftBatch,
            temperature=0.35,
        )
        for opp in batch.opportunities:
            normalized = opp.model_copy(update={"opportunity_id": f"OPP-{counter:03d}"})
            all_opps.append(normalized)
            counter += 1

    s1_artifact = Stage1Artifact(run_id=run_id, domain=domain, opportunities=all_opps)
    write_json(run_dir / "01_opportunities.json", s1_artifact)
    validate_stage_1(run_dir, run_id)
    print(f"  [GATE PASS] Saved & verified {len(s1_artifact.opportunities)} opportunities.")
    return s1_artifact


def run_stage_2(
    client: OllamaStructuredClient,
    worker_model: str,
    critic_model: str,
    run_dir: Path,
    run_id: str,
    s1: Stage1Artifact,
    constraints: str,
    max_selected: int,
    max_critic_loops: int,
) -> Stage2Artifact:
    print(f"\n=== STAGE 2: First-Principles & Contradiction Analysis ({len(s1.opportunities)} items) ===")
    raw_analyses: list[OpportunityAnalysis] = []

    for idx, opp in enumerate(s1.opportunities, start=1):
        anl_id = f"ANL-{idx:03d}"
        print(f"  -> Analyzing {opp.opportunity_id} -> {anl_id} ({opp.title})...")

        base_prompt = (
            f"Domain: {s1.domain}\nConstraints: {constraints}\n\n"
            f"Analyze this Stage 1 Opportunity:\n{opp.model_dump_json(indent=2)}\n\n"
            f"Requirements:\n"
            f"- Set analysis_id='{anl_id}' and opportunity_id='{opp.opportunity_id}'.\n"
            f"- Decompose into First-Principles fundamental truths, challenged assumptions, and >=3-step cause-effect chain.\n"
            f"- Formulate both a Technical Contradiction and a Physical Contradiction (opposite states on the SAME control_element).\n"
            f"- Compute ICE score strictly: composite_score = round((impact * confidence * ease) / 10.0, 2).\n"
            f"- Set selected_for_stage_3=False initially (ranking selection happens deterministically after all analyses complete)."
        )

        analysis = client.generate_structured(
            model=worker_model,
            system_prompt=WORKER_SYSTEM_PROMPT,
            user_prompt=base_prompt,
            schema_cls=OpportunityAnalysis,
        )

        # Adversarial Critic Audit Loop
        for loop in range(1, max_critic_loops + 1):
            verdict = client.generate_structured(
                model=critic_model,
                system_prompt=CRITIC_SYSTEM_PROMPT,
                user_prompt=(
                    f"Audit Stage 2 Analysis `{anl_id}` for `{opp.opportunity_id}`:\n"
                    f"Opportunity:\n{opp.model_dump_json(indent=2)}\n\n"
                    f"Candidate Analysis:\n{analysis.model_dump_json(indent=2)}"
                ),
                schema_cls=CriticVerdict,
                temperature=0.1,
            )
            if verdict.status == "PASS":
                print(f"     [Critic PASS] {anl_id} verified on pass {loop}.")
                break

            print(f"     [Critic REJECT {loop}/{max_critic_loops}] Defects: {verdict.blocking_defects}")
            revision_prompt = (
                f"{base_prompt}\n\n"
                f"Previous Draft:\n{analysis.model_dump_json(indent=2)}\n\n"
                f"Adversarial Verifier Rejected with Defects:\n"
                + "\n".join(f"- {d}" for d in verdict.blocking_defects)
                + f"\nRevision Instructions: {verdict.revision_instructions}"
            )
            analysis = client.generate_structured(
                model=worker_model,
                system_prompt=WORKER_SYSTEM_PROMPT,
                user_prompt=revision_prompt,
                schema_cls=OpportunityAnalysis,
            )

        # Enforce exact foreign keys & exact ICE composite math before funnel ranking
        ice = analysis.ice_score
        exact_composite = round((ice.impact * ice.confidence * ice.ease) / 10.0, 2)
        fixed_ice = ice.model_copy(update={"composite_score": exact_composite})
        analysis = analysis.model_copy(
            update={
                "analysis_id": anl_id,
                "opportunity_id": opp.opportunity_id,
                "ice_score": fixed_ice,
                "selected_for_stage_3": False,
            }
        )
        raw_analyses.append(analysis)

    # Deterministic Funnel Pruning Selection: select top `max_selected` by ICE composite_score
    ranked = sorted(
        raw_analyses,
        key=lambda a: (-a.ice_score.composite_score, a.opportunity_id),
    )
    selected_opp_ids = {a.opportunity_id for a in ranked[:max_selected]}

    final_analyses = [
        a.model_copy(update={"selected_for_stage_3": (a.opportunity_id in selected_opp_ids)})
        for a in raw_analyses
    ]

    s2_artifact = Stage2Artifact(run_id=run_id, analyses=final_analyses)
    write_json(run_dir / "02_analyses.json", s2_artifact)
    validate_stage_2(run_dir, run_id, max_selected=max_selected, enforce_rank_order=True)
    print(f"  [GATE PASS] Stage 2 verified; top {len(selected_opp_ids)} selected: {sorted(selected_opp_ids)}")
    return s2_artifact


def run_stage_3(
    client: OllamaStructuredClient,
    worker_model: str,
    critic_model: str,
    run_dir: Path,
    run_id: str,
    s1: Stage1Artifact,
    s2: Stage2Artifact,
    constraints: str,
    max_selected: int,
    max_critic_loops: int,
) -> Stage3Artifact:
    selected_analyses = [a for a in s2.analyses if a.selected_for_stage_3]
    opp_by_id = {o.opportunity_id: o for o in s1.opportunities}
    print(f"\n=== STAGE 3: IFR-Aligned Recommendations ({len(selected_analyses)} selected) ===")

    recommendations: list[Recommendation] = []
    for idx, anl in enumerate(selected_analyses, start=1):
        rec_id = f"REC-{idx:03d}"
        opp = opp_by_id[anl.opportunity_id]
        print(f"  -> Synthesizing {rec_id} for {opp.opportunity_id} / {anl.analysis_id}...")

        base_prompt = (
            f"Domain: {s1.domain}\nConstraints: {constraints}\n\n"
            f"Opportunity:\n{opp.model_dump_json(indent=2)}\n\n"
            f"Verified Stage 2 Analysis:\n{anl.model_dump_json(indent=2)}\n\n"
            f"Requirements:\n"
            f"- Set recommendation_id='{rec_id}', opportunity_id='{opp.opportunity_id}', analysis_id='{anl.analysis_id}'.\n"
            f"- Define an Ideal Final Result (IFR) where the system solves the bottleneck ITSELF using free existing resources.\n"
            f"- Eliminate the Stage 2 Physical/Technical contradiction via an explicit TRIZ separation_strategy and named TRIZ principles—do NOT propose a weak middle-ground compromise.\n"
            f"- Provide concrete trade_offs_and_risks with mitigations and a quantitative falsifiable_validation_metric."
        )

        rec = client.generate_structured(
            model=worker_model,
            system_prompt=WORKER_SYSTEM_PROMPT,
            user_prompt=base_prompt,
            schema_cls=Recommendation,
        )

        for loop in range(1, max_critic_loops + 1):
            verdict = client.generate_structured(
                model=critic_model,
                system_prompt=CRITIC_SYSTEM_PROMPT,
                user_prompt=(
                    f"Audit Stage 3 Recommendation `{rec_id}` against Stage 2 Contradictions & IFR rules:\n"
                    f"Stage 2 Analysis:\n{anl.model_dump_json(indent=2)}\n\n"
                    f"Candidate Recommendation:\n{rec.model_dump_json(indent=2)}"
                ),
                schema_cls=CriticVerdict,
                temperature=0.1,
            )
            if verdict.status == "PASS":
                print(f"     [Critic PASS] {rec_id} verified on pass {loop}.")
                break

            print(f"     [Critic REJECT {loop}/{max_critic_loops}] Defects: {verdict.blocking_defects}")
            revision_prompt = (
                f"{base_prompt}\n\n"
                f"Previous Draft:\n{rec.model_dump_json(indent=2)}\n\n"
                f"Adversarial Verifier Rejected with Defects:\n"
                + "\n".join(f"- {d}" for d in verdict.blocking_defects)
                + f"\nRevision Instructions: {verdict.revision_instructions}"
            )
            rec = client.generate_structured(
                model=worker_model,
                system_prompt=WORKER_SYSTEM_PROMPT,
                user_prompt=revision_prompt,
                schema_cls=Recommendation,
            )

        rec = rec.model_copy(
            update={
                "recommendation_id": rec_id,
                "opportunity_id": opp.opportunity_id,
                "analysis_id": anl.analysis_id,
            }
        )
        recommendations.append(rec)

    s3_artifact = Stage3Artifact(run_id=run_id, recommendations=recommendations)
    write_json(run_dir / "03_recommendations.json", s3_artifact)
    validate_stage_3(run_dir, run_id, max_selected=max_selected, enforce_rank_order=True)
    print(f"  [GATE PASS] Stage 3 verified ({len(recommendations)} recommendations).")
    return s3_artifact


def run_stage_4(
    client: OllamaStructuredClient,
    architect_model: str,
    critic_model: str,
    run_dir: Path,
    run_id: str,
    s1: Stage1Artifact,
    s2: Stage2Artifact,
    s3: Stage3Artifact,
    constraints: str,
    max_selected: int,
    max_critic_loops: int,
) -> Stage4Artifact:
    opp_by_id = {o.opportunity_id: o for o in s1.opportunities}
    anl_by_id = {a.analysis_id: a for a in s2.analyses}
    print(f"\n=== STAGE 4: Implementation Blueprints ({len(s3.recommendations)} recommendations) ===")

    blueprints: list[ImplementationBlueprint] = []
    for idx, rec in enumerate(s3.recommendations, start=1):
        imp_id = f"IMP-{idx:03d}"
        opp = opp_by_id[rec.opportunity_id]
        anl = anl_by_id[rec.analysis_id]
        print(f"  -> Architecting {imp_id} for {rec.recommendation_id} ({rec.title})...")

        base_prompt = (
            f"Domain: {s1.domain}\nConstraints: {constraints}\n\n"
            f"Opportunity:\n{opp.model_dump_json(indent=2)}\n\n"
            f"Root Cause & Contradictions:\n{anl.model_dump_json(indent=2)}\n\n"
            f"Approved IFR Recommendation:\n{rec.model_dump_json(indent=2)}\n\n"
            f"Requirements:\n"
            f"- Set blueprint_id='{imp_id}', recommendation_id='{rec.recommendation_id}', opportunity_id='{rec.opportunity_id}'.\n"
            f"- Preserve the Ideal Final Result (zero unnecessary bloat or CapEx violations).\n"
            f"- Define >= 2 sequential phases (phase_number 1, 2, ...) with concrete deliverables, machine-verifiable verification_gate, and quantitative kill_switch_criteria."
        )

        bp = client.generate_structured(
            model=architect_model,
            system_prompt=WORKER_SYSTEM_PROMPT,
            user_prompt=base_prompt,
            schema_cls=ImplementationBlueprint,
        )

        for loop in range(1, max_critic_loops + 1):
            verdict = client.generate_structured(
                model=critic_model,
                system_prompt=CRITIC_SYSTEM_PROMPT,
                user_prompt=(
                    f"Audit Stage 4 Implementation Blueprint `{imp_id}` against IFR & kill-switch rules:\n"
                    f"Stage 3 Recommendation:\n{rec.model_dump_json(indent=2)}\n\n"
                    f"Candidate Blueprint:\n{bp.model_dump_json(indent=2)}"
                ),
                schema_cls=CriticVerdict,
                temperature=0.1,
            )
            if verdict.status == "PASS":
                print(f"     [Critic PASS] {imp_id} verified on pass {loop}.")
                break

            print(f"     [Critic REJECT {loop}/{max_critic_loops}] Defects: {verdict.blocking_defects}")
            revision_prompt = (
                f"{base_prompt}\n\n"
                f"Previous Draft:\n{bp.model_dump_json(indent=2)}\n\n"
                f"Adversarial Verifier Rejected with Defects:\n"
                + "\n".join(f"- {d}" for d in verdict.blocking_defects)
                + f"\nRevision Instructions: {verdict.revision_instructions}"
            )
            bp = client.generate_structured(
                model=architect_model,
                system_prompt=WORKER_SYSTEM_PROMPT,
                user_prompt=revision_prompt,
                schema_cls=ImplementationBlueprint,
            )

        # Normalize sequential phase numbering & exact foreign keys
        normalized_phases = [
            ph.model_copy(update={"phase_number": p_idx})
            for p_idx, ph in enumerate(bp.phases, start=1)
        ]
        bp = bp.model_copy(
            update={
                "blueprint_id": imp_id,
                "recommendation_id": rec.recommendation_id,
                "opportunity_id": rec.opportunity_id,
                "phases": normalized_phases,
            }
        )
        blueprints.append(bp)

    s4_artifact = Stage4Artifact(run_id=run_id, blueprints=blueprints)
    write_json(run_dir / "04_implementations.json", s4_artifact)
    validate_stage_4(run_dir, run_id, max_selected=max_selected, enforce_rank_order=True)
    print(f"  [GATE PASS] Stage 4 verified ({len(blueprints)} blueprints).")
    return s4_artifact


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run the 4-stage Pydantic-gated innovation pipeline locally via Ollama."
    )
    parser.add_argument("--run-id", required=True, help="Unique run ID (e.g. 2026-11-13_marina).")
    parser.add_argument("--domain", required=True, help="Target industry or technical domain.")
    parser.add_argument(
        "--user-groups",
        required=True,
        help="Comma-separated list of target user groups to scout in Stage 1.",
    )
    parser.add_argument(
        "--constraints",
        default="Minimize CapEx and operational complexity; favor passive or existing-resource solutions.",
        help="Hard engineering, budget, or operational constraints.",
    )
    parser.add_argument(
        "--opps-per-group",
        type=int,
        default=3,
        help="Number of opportunities to discover per user group in Stage 1 (default: 3).",
    )
    parser.add_argument(
        "--max-selected",
        type=int,
        default=3,
        help="Funnel pruning cap for Stage 2 -> Stage 3 advancement (default: 3).",
    )
    parser.add_argument(
        "--worker-model",
        default=os.environ.get("OLLAMA_WORKER_MODEL", "granite4:latest"),
        help="Local Ollama model for Stages 1-3 generation (default: granite4:latest).",
    )
    parser.add_argument(
        "--critic-model",
        default=os.environ.get("OLLAMA_CRITIC_MODEL", "nemotron-3:30b"),
        help="Local Ollama model for Adversarial Critic & Stage 4 architecture (default: nemotron-3:30b).",
    )
    parser.add_argument(
        "--ollama-url",
        default=os.environ.get("OLLAMA_HOST", "http://localhost:11434"),
        help="Ollama server URL (default: http://localhost:11434).",
    )
    parser.add_argument(
        "--max-critic-loops",
        type=int,
        default=2,
        help="Max adversarial critic revision cycles per item (default: 2).",
    )
    parser.add_argument(
        "--artifacts-root",
        type=Path,
        default=REPO_ROOT / "artifacts",
        help="Root directory for run artifacts (default: ./artifacts).",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    run_dir: Path = args.artifacts_root / args.run_id
    user_groups = [g.strip() for g in args.user_groups.split(",") if g.strip()]

    if not user_groups:
        print("[ERROR] --user-groups must contain at least one non-empty group.", file=sys.stderr)
        return 2

    client = OllamaStructuredClient(base_url=args.ollama_url)

    try:
        s1 = run_stage_1(
            client=client,
            worker_model=args.worker_model,
            run_dir=run_dir,
            run_id=args.run_id,
            domain=args.domain,
            user_groups=user_groups,
            constraints=args.constraints,
            opps_per_group=args.opps_per_group,
        )
        s2 = run_stage_2(
            client=client,
            worker_model=args.worker_model,
            critic_model=args.critic_model,
            run_dir=run_dir,
            run_id=args.run_id,
            s1=s1,
            constraints=args.constraints,
            max_selected=args.max_selected,
            max_critic_loops=args.max_critic_loops,
        )
        s3 = run_stage_3(
            client=client,
            worker_model=args.worker_model,
            critic_model=args.critic_model,
            run_dir=run_dir,
            run_id=args.run_id,
            s1=s1,
            s2=s2,
            constraints=args.constraints,
            max_selected=args.max_selected,
            max_critic_loops=args.max_critic_loops,
        )
        s4 = run_stage_4(
            client=client,
            architect_model=args.critic_model,
            critic_model=args.critic_model,
            run_dir=run_dir,
            run_id=args.run_id,
            s1=s1,
            s2=s2,
            s3=s3,
            constraints=args.constraints,
            max_selected=args.max_selected,
            max_critic_loops=args.max_critic_loops,
        )

        # Final transitive gate + Markdown dossier compilation
        validate_stage_4(run_dir, args.run_id, max_selected=args.max_selected, enforce_rank_order=True)
        report_md = build_markdown_dossier(s1, s2, s3, s4)
        report_path = run_dir / "FINAL_REPORT.md"
        report_path.write_text(report_md, encoding="utf-8")

        print(f"\n[SUCCESS] Pipeline complete for '{args.run_id}'.")
        print(f"          Final Dossier: {report_path}")
        return 0

    except (RuntimeError, ValidationError, GateFailure) as exc:
        print(f"\n[ABORT] Pipeline halted:\n{exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

Key Local Execution Features

• Zero External Dependencies Beyond Pydantic: Uses Python's standard library urllib.request to call Ollama's native /api/chat endpoint with format=Model.model_json_schema(), ensuring grammar-constrained JSON output without requiring extra SDK packages.

• Two-Tier Self-Correction:

  1. Schema Repair Loop (generate_structured): If an Ollama model violates a Pydantic v2 constraint (such as extra="forbid", regex ID format, or identical physical contradiction states), the exact ValidationError is fed back to the model for up to 3 repair passes before advancing.

  2. Adversarial Critic Loop (CriticVerdict): Mirrors /swarm's Verifier role by having your heavier reasoning model (nemotron-3:30b by default) audit every Stage 2, 3, and 4 candidate against First-Principles, Contradiction, and IFR rules before writing the stage artifact.

• Deterministic Funnel & Math Normalization: Enforces exact ICE arithmetic (round((impact * confidence * ease) / 10.0, 2)) and ranks candidates deterministically in Python so rank inversions cannot occur during Stage 2 pruning.

Write tests/test_pipeline.py using pytest and tmp_path to test schemas/pipeline.py, scripts/validate_stage.py (catching FK violations, lineage mismatches, ICE rank inversions, and pruned opportunity leaks), and scripts/render_report.py.

Below is the complete tests/test_pipeline.py suite built with pytest and tmp_path. It tests positive end-to-end validation and report compilation alongside every negative schema rule, foreign-key gate, funnel pruning constraint, and CLI exit code across schemas/pipeline.py, scripts/validate_stage.py, and scripts/render_report.py.

tests/test_pipeline.py

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest
from pydantic import ValidationError

REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from schemas.pipeline import (  # noqa: E402
    ContradictionResolution,
    FirstPrinciplesDecomposition,
    ICEScore,
    IdealFinalResultSpec,
    ImplementationBlueprint,
    ImplementationPhase,
    Opportunity,
    OpportunityAnalysis,
    PhysicalContradiction,
    Recommendation,
    Stage1Artifact,
    Stage2Artifact,
    Stage3Artifact,
    Stage4Artifact,
    TechnicalContradiction,
    TradeOffEntry,
)
from scripts.render_report import build_markdown_dossier  # noqa: E402
from scripts.validate_stage import (  # noqa: E402
    GateFailure,
    validate_stage_1,
    validate_stage_2,
    validate_stage_3,
    validate_stage_4,
)


# ==============================================================================
# Fixtures & Deterministic Valid Dataset Builders
# ==============================================================================

RUN_ID = "2026-11-13_test_run"


def make_opportunity(idx: int) -> Opportunity:
    return Opportunity(
        opportunity_id=f"OPP-{idx:03d}",
        title=f"Automated Thermal Drift Compensation #{idx}",
        industry_segment="Precision Optical Manufacturing",
        target_user_group="Cleanroom Metrology Technicians",
        operational_bottleneck=(
            f"Technicians manually recalibrate interferometer stages every 45 minutes due to ambient HVAC cycling (#{idx})."
        ),
        baseline_friction_metric="18 min downtime per shift; 12 micron thermal drift variance",
        supporting_evidence=[
            "Q3 metrology shift logs showing 4.2 hours/week lost to warm-up re-zeroing",
            "Thermocouple trace correlating 1.4C HVAC swing with baseline shift",
        ],
    )


def make_analysis(
    idx: int,
    *,
    impact: float,
    confidence: float,
    ease: float,
    selected: bool,
    opp_id: str | None = None,
) -> OpportunityAnalysis:
    composite = round((impact * confidence * ease) / 10.0, 2)
    return OpportunityAnalysis(
        analysis_id=f"ANL-{idx:03d}",
        opportunity_id=opp_id or f"OPP-{idx:03d}",
        first_principles=FirstPrinciplesDecomposition(
            fundamental_truths=[
                "Linear thermal expansion delta_L = alpha * L * delta_T is deterministic for known alloys.",
                "Optical phase error scales linearly with uncompensated path-length change.",
            ],
            conventional_assumptions_challenged=[
                "Assumed active Peltier enclosure cooling is required rather than passive phase-error cancellation."
            ],
            cause_effect_chain=[
                "HVAC compressor cycles every 22 minutes causing 1.4C air gradient.",
                "Granite and aluminum optical rail expand at unequal coefficients (alpha mismatch).",
                "Interferometer reference arm shifts relative to measurement arm.",
                "Technician halts inspection run to perform manual zero-offset calibration.",
            ],
            root_cause_statement="Differential thermal expansion between optical rail materials without real-time reference path compensation.",
        ),
        technical_contradiction=TechnicalContradiction(
            improving_parameter="Measurement positional stability across temperature swings",
            worsening_parameter="System mass, power draw, and enclosure complexity",
            conflict_mechanism="Adding active thermal chambers stabilizes temperature but triples footprint and power consumption.",
        ),
        physical_contradiction=PhysicalContradiction(
            control_element="Optical reference path length",
            state_a_requirement="Must remain invariant during active part inspection",
            state_a_justification="Prevents phase drift from corrupting sub-micron surface measurements",
            state_b_requirement="Must vary dynamically with ambient thermal expansion",
            state_b_justification="Absorbs structural rail expansion without mechanical actuators",
        ),
        ice_score=ICEScore(
            impact=impact,
            confidence=confidence,
            ease=ease,
            composite_score=composite,
            scoring_rationale="High verified downtime savings using existing telemetry channels and low-cost RTD sensors.",
        ),
        selected_for_stage_3=selected,
    )


def make_recommendation(
    idx: int,
    *,
    opp_id: str,
    anl_id: str,
) -> Recommendation:
    return Recommendation(
        recommendation_id=f"REC-{idx:03d}",
        opportunity_id=opp_id,
        analysis_id=anl_id,
        title=f"Self-Compensating Differential Reference Path #{idx}",
        ifr_alignment=IdealFinalResultSpec(
            ifr_statement="The optical rail itself cancels thermal phase drift during measurement without external heaters or coolers.",
            free_existing_resources_used=[
                "Existing unused secondary photodiode quadrature channel",
                "Ambient cleanroom airflow thermal gradient already present",
            ],
            eliminated_harms=[
                "18 min/shift manual recalibration downtime",
                "Active chiller vibration noise",
            ],
            ideality_gain_mechanism="Replaces active thermal stabilization hardware with differential phase subtraction using existing sensor channels.",
        ),
        contradiction_resolution=ContradictionResolution(
            separation_strategy="separation_between_parts_and_whole",
            triz_principles_applied=[
                "Principle 2: Taking Out",
                "Principle 40: Composite Materials",
            ],
            how_contradiction_is_eliminated=(
                "Pairs an Invar reference rod (invariant part) alongside the aluminum rail (expanding whole) so differential optical phase directly measures and subtracts thermal expansion in real time."
            ),
        ),
        mechanism_of_action=(
            "Mount an unrestrained Invar reference rod anchored at one datum point; route the secondary quadrature beam off its end reflector to compute real-time thermal delta_L and subtract it in firmware."
        ),
        trade_offs_and_risks=[
            TradeOffEntry(
                risk_or_cost_factor="Differential humidity absorption on open optical path",
                severity="low",
                mitigation_mechanism="Apply Edlen refractive index correction using existing cleanroom RH sensor.",
            )
        ],
        falsifiable_validation_metric="Reduce thermal drift error from 12 microns to < 0.8 microns over a 2.0C swing with zero manual stops",
    )


def make_blueprint(
    idx: int,
    *,
    rec_id: str,
    opp_id: str,
) -> ImplementationBlueprint:
    return ImplementationBlueprint(
        blueprint_id=f"IMP-{idx:03d}",
        recommendation_id=rec_id,
        opportunity_id=opp_id,
        architecture_summary="Passive Invar datum rod coupled to existing quadrature firmware subtraction loop.",
        bill_of_materials_or_stack=[
            "1x 400mm Invar-36 ground reference rod ($140)",
            "1x Kinematic datum flexure mount ($65)",
            "Firmware phase-subtraction filter module (Python/C++)",
        ],
        phases=[
            ImplementationPhase(
                phase_number=1,
                phase_name="Datum Fixture & Quadrature Tap",
                deliverables=[
                    "Install kinematic Invar rod on bench #1",
                    "Capture 24-hour dual-channel drift telemetry CSV",
                ],
                verification_gate="Correlation R^2 >= 0.98 between differential optical phase and laser tracker ground truth",
                kill_switch_criteria="Abort if mechanical mount hysteresis exceeds 0.5 microns across 3 thermal cycles",
            ),
            ImplementationPhase(
                phase_number=2,
                phase_name="Closed-Loop Firmware Compensation",
                deliverables=[
                    "Deploy real-time subtraction filter in acquisition pipeline",
                    "Run 5-shift production validation without manual zeroing",
                ],
                verification_gate="Peak-to-peak drift < 0.8 microns across full shift with exit code 0 in pytest suite",
                kill_switch_criteria="Rollback to baseline firmware if latency exceeds 2ms per frame or error exceeds 1.0 micron",
            ),
        ],
    )


@pytest.fixture
def populated_run_dir(tmp_path: Path) -> Path:
    """Create a fully valid Stage 1–4 artifact directory with 3 opportunities (top 2 selected)."""
    run_dir = tmp_path / RUN_ID
    run_dir.mkdir(parents=True, exist_ok=True)

    s1 = Stage1Artifact(
        run_id=RUN_ID,
        domain="Precision Optical Manufacturing",
        opportunities=[make_opportunity(1), make_opportunity(2), make_opportunity(3)],
    )

    # OPP-001 (ICE 72.0) and OPP-002 (ICE 54.0) selected; OPP-003 (ICE 24.0) pruned
    s2 = Stage2Artifact(
        run_id=RUN_ID,
        analyses=[
            make_analysis(1, impact=9.0, confidence=8.0, ease=10.0, selected=True),
            make_analysis(2, impact=9.0, confidence=6.0, ease=10.0, selected=True),
            make_analysis(3, impact=6.0, confidence=5.0, ease=8.0, selected=False),
        ],
    )

    s3 = Stage3Artifact(
        run_id=RUN_ID,
        recommendations=[
            make_recommendation(1, opp_id="OPP-001", anl_id="ANL-001"),
            make_recommendation(2, opp_id="OPP-002", anl_id="ANL-002"),
        ],
    )

    s4 = Stage4Artifact(
        run_id=RUN_ID,
        blueprints=[
            make_blueprint(1, rec_id="REC-001", opp_id="OPP-001"),
            make_blueprint(2, rec_id="REC-002", opp_id="OPP-002"),
        ],
    )

    (run_dir / "01_opportunities.json").write_text(s1.model_dump_json(indent=2), encoding="utf-8")
    (run_dir / "02_analyses.json").write_text(s2.model_dump_json(indent=2), encoding="utf-8")
    (run_dir / "03_recommendations.json").write_text(s3.model_dump_json(indent=2), encoding="utf-8")
    (run_dir / "04_implementations.json").write_text(s4.model_dump_json(indent=2), encoding="utf-8")
    return run_dir


# ==============================================================================
# 1. Pydantic v2 Schema Unit Tests (`schemas/pipeline.py`)
# ==============================================================================

def test_schema_forbids_extra_fields() -> None:
    opp_dict = make_opportunity(1).model_dump()
    opp_dict["unexpected_llm_field"] = "should be rejected"
    with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
        Opportunity.model_validate(opp_dict)


def test_schema_enforces_strict_types() -> None:
    opp_dict = make_opportunity(1).model_dump()
    opp_dict["supporting_evidence"] = "single string instead of list"
    with pytest.raises(ValidationError):
        Opportunity.model_validate(opp_dict)


def test_schema_rejects_bad_id_patterns() -> None:
    opp_dict = make_opportunity(1).model_dump()
    opp_dict["opportunity_id"] = "OPP-1"  # Must be 3 digits
    with pytest.raises(ValidationError):
        Opportunity.model_validate(opp_dict)


def test_schema_rejects_identical_physical_contradiction_states() -> None:
    with pytest.raises(ValidationError, match="genuinely opposing states"):
        PhysicalContradiction(
            control_element="Valve aperture",
            state_a_requirement="Must remain fully closed under load",
            state_a_justification="Prevents backflow",
            state_b_requirement="must remain fully closed under load",
            state_b_justification="Identical state disguised as contradiction",
        )


def test_schema_rejects_invalid_ice_composite_math() -> None:
    with pytest.raises(ValidationError, match="ICE composite_score"):
        ICEScore(
            impact=8.0,
            confidence=8.0,
            ease=8.0,
            composite_score=99.0,  # Expected (8*8*8)/10 = 51.2
            scoring_rationale="Inflated composite score that violates deterministic math.",
        )


def test_schema_rejects_non_sequential_implementation_phases() -> None:
    bp = make_blueprint(1, rec_id="REC-001", opp_id="OPP-001")
    bad_phases = [
        bp.phases[0].model_copy(update={"phase_number": 1}),
        bp.phases[1].model_copy(update={"phase_number": 3}),  # Skipped phase 2
    ]
    with pytest.raises(ValidationError, match="sequential"):
        ImplementationBlueprint.model_validate(
            {**bp.model_dump(), "phases": [p.model_dump() for p in bad_phases]}
        )


# ==============================================================================
# 2. Cross-Stage Foreign-Key & Funnel Gate Tests (`scripts/validate_stage.py`)
# ==============================================================================

def test_validate_stage_4_happy_path(populated_run_dir: Path) -> None:
    s1, s2, s3, s4 = validate_stage_4(
        artifacts_dir=populated_run_dir,
        run_id=RUN_ID,
        max_selected=2,
        enforce_rank_order=True,
    )
    assert len(s1.opportunities) == 3
    assert sum(1 for a in s2.analyses if a.selected_for_stage_3) == 2
    assert len(s3.recommendations) == 2
    assert len(s4.blueprints) == 2


def test_stage_2_catches_missing_and_phantom_foreign_keys(populated_run_dir: Path) -> None:
    s2_path = populated_run_dir / "02_analyses.json"
    s2_data = json.loads(s2_path.read_text(encoding="utf-8"))
    # Replace OPP-003 analysis with phantom OPP-099
    s2_data["analyses"][2]["opportunity_id"] = "OPP-099"
    s2_path.write_text(json.dumps(s2_data, indent=2), encoding="utf-8")

    with pytest.raises(GateFailure) as exc_info:
        validate_stage_2(populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=True)

    msg = str(exc_info.value)
    assert "Missing analyses for Stage 1 opportunities: ['OPP-003']" in msg
    assert "non-existent Stage 1 opportunities: ['OPP-099']" in msg


def test_stage_2_catches_funnel_cardinality_exceeded(populated_run_dir: Path) -> None:
    with pytest.raises(GateFailure, match="exceeding cap of 1"):
        validate_stage_2(populated_run_dir, RUN_ID, max_selected=1, enforce_rank_order=True)


def test_stage_2_catches_ice_rank_inversion(populated_run_dir: Path) -> None:
    s2_path = populated_run_dir / "02_analyses.json"
    s2_data = json.loads(s2_path.read_text(encoding="utf-8"))
    # Deselect OPP-002 (ICE 54.0) and select OPP-003 (ICE 24.0) -> rank inversion
    s2_data["analyses"][1]["selected_for_stage_3"] = False
    s2_data["analyses"][2]["selected_for_stage_3"] = True
    s2_path.write_text(json.dumps(s2_data, indent=2), encoding="utf-8")

    with pytest.raises(GateFailure, match="Stage 2 Rank Inversion"):
        validate_stage_2(populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=True)

    # Verify --allow-rank-override flag permits explicit strategic override
    _, s2_override = validate_stage_2(
        populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=False
    )
    assert [a.opportunity_id for a in s2_override.analyses if a.selected_for_stage_3] == [
        "OPP-001",
        "OPP-003",
    ]


def test_stage_3_catches_pruned_opportunity_leak(populated_run_dir: Path) -> None:
    s3_path = populated_run_dir / "03_recommendations.json"
    s3_data = json.loads(s3_path.read_text(encoding="utf-8"))
    # Append recommendation targeting pruned OPP-003 / ANL-003
    leaked_rec = make_recommendation(3, opp_id="OPP-003", anl_id="ANL-003").model_dump()
    s3_data["recommendations"].append(leaked_rec)
    s3_path.write_text(json.dumps(s3_data, indent=2), encoding="utf-8")

    with pytest.raises(GateFailure, match="Stage 3 Funnel Leak"):
        validate_stage_3(populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=True)


def test_stage_3_catches_lineage_pair_mismatch(populated_run_dir: Path) -> None:
    s3_path = populated_run_dir / "03_recommendations.json"
    s3_data = json.loads(s3_path.read_text(encoding="utf-8"))
    # Pair OPP-001 with ANL-002 (which belongs to OPP-002)
    s3_data["recommendations"][0]["analysis_id"] = "ANL-002"
    s3_path.write_text(json.dumps(s3_data, indent=2), encoding="utf-8")

    with pytest.raises(GateFailure, match="Stage 3 Lineage Mismatch"):
        validate_stage_3(populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=True)


def test_stage_3_catches_uncovered_selected_opportunity(populated_run_dir: Path) -> None:
    s3_path = populated_run_dir / "03_recommendations.json"
    s3_data = json.loads(s3_path.read_text(encoding="utf-8"))
    # Drop recommendation for selected OPP-002
    s3_data["recommendations"] = [s3_data["recommendations"][0]]
    s3_path.write_text(json.dumps(s3_data, indent=2), encoding="utf-8")

    with pytest.raises(GateFailure, match="missing recommendations: \\['OPP-002'\\]"):
        validate_stage_3(populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=True)


def test_stage_4_catches_blueprint_lineage_mismatch(populated_run_dir: Path) -> None:
    s4_path = populated_run_dir / "04_implementations.json"
    s4_data = json.loads(s4_path.read_text(encoding="utf-8"))
    # Pair REC-001 with OPP-002
    s4_data["blueprints"][0]["opportunity_id"] = "OPP-002"
    s4_path.write_text(json.dumps(s4_data, indent=2), encoding="utf-8")

    with pytest.raises(GateFailure, match="Stage 4 Lineage Mismatch"):
        validate_stage_4(populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=True)


# ==============================================================================
# 3. Markdown Report Compiler & CLI Subprocess Tests (`scripts/render_report.py`)
# ==============================================================================

def test_render_report_generates_complete_traceable_dossier(populated_run_dir: Path) -> None:
    s1, s2, s3, s4 = validate_stage_4(
        populated_run_dir, RUN_ID, max_selected=2, enforce_rank_order=True
    )
    md = build_markdown_dossier(s1, s2, s3, s4)

    assert f"# Innovation Discovery & Implementation Dossier — `{RUN_ID}`" in md
    assert "3 Discovered → 3 Analyzed → 2 Selected → 2 Recommendations → 2 Implementation Blueprints" in md
    assert "`OPP-001` → `ANL-001` → `REC-001`" in md
    assert "Stage 4: Implementation Blueprint (`IMP-001` → `REC-001`)" in md
    # Verify pruned candidate appears in Appendix and not in Selected Deep-Dive
    assert "## 3. Appendix — Pruned Opportunities Audit Log" in md
    assert "`OPP-003` / `ANL-003`" in md


def test_cli_end_to_end_exit_codes(populated_run_dir: Path) -> None:
    artifacts_root = populated_run_dir.parent

    # 1. Valid CLI invocation returns exit code 0 and writes FINAL_REPORT.md
    res_ok = subprocess.run(
        [
            sys.executable,
            str(REPO_ROOT / "scripts" / "render_report.py"),
            "--run-id",
            RUN_ID,
            "--artifacts-root",
            str(artifacts_root),
            "--max-selected",
            "2",
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    assert res_ok.returncode == 0, res_ok.stderr
    assert (populated_run_dir / "FINAL_REPORT.md").is_file()

    # 2. Missing run-id returns exit code 2
    res_missing = subprocess.run(
        [
            sys.executable,
            str(REPO_ROOT / "scripts" / "validate_stage.py"),
            "--run-id",
            "nonexistent-run",
            "--stage",
            "1",
            "--artifacts-root",
            str(artifacts_root),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    assert res_missing.returncode == 2

Running the Verification Suite

With schemas/pipeline.py, scripts/validate_stage.py, scripts/render_report.py, and tests/test_pipeline.py in place, you can run your canonical verification commands directly:

pytest -v --tb=short
mypy --strict schemas scripts tests
ruff check schemas scripts tests

What is the pyproject.toml and README.md?

Below are the complete pyproject.toml and README.md files configured to tie together your Pydantic v2 schemas (schemas/pipeline.py), deterministic validation gates (scripts/validate_stage.py), Markdown dossier compiler (scripts/render_report.py), local Ollama runner (scripts/run_local_ollama.py), and Google Antigravity /swarm workflow.

1. pyproject.toml

Configures Python 3.12+, Pydantic v2 (>=2.9.0), mypy --strict with the pydantic.mypy plugin, ruff linting/formatting, and pytest discovery so both local developers and Antigravity /swarm Verifier agents run identical checks.

Ini, TOML

[build-system]
requires = ["setuptools>=69.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "innovation-swarm-engine"
version = "0.1.0"
description = "Schema-gated, First-Principles and TRIZ/IFR 4-stage innovation pipeline for Google Antigravity /swarm and local Ollama models."
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "pydantic>=2.9.0",
]

[project.optional-dependencies]
dev = [
    "mypy>=1.11.0",
    "pytest>=8.3.0",
    "ruff>=0.6.0",
]

[tool.setuptools.packages.find]
where = ["."]
include = ["schemas*", "scripts*"]

[tool.pytest.ini_options]
minversion = "8.0"
testpaths = ["tests"]
pythonpath = ["."]
addopts = "-v --tb=short --strict-markers"

[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true

[tool.pydantic-mypy]
init_forbid_extra = true
init_typed = true
warn_required_dynamic_aliases = true

[tool.ruff]
target-version = "py312"
line-length = 100
src = ["schemas", "scripts", "tests"]

[tool.ruff.lint]
select = [
    "E",   # pycodestyle errors
    "W",   # pycodestyle warnings
    "F",   # pyflakes
    "I",   # isort
    "B",   # flake8-bugbear
    "UP",  # pyupgrade
    "SIM", # flake8-simplify
    "RUF", # Ruff-specific rules
]
ignore = [
    "E501", # line length handled by ruff format
]

[tool.ruff.format]
quote-style = "double"
indent-style = "space"

2. README.md

Documents repository setup, verification commands, both execution modes (Antigravity /swarm and local Ollama CLI), and the stage-by-stage contract architecture.

# Innovation Swarm Engine

A deterministic, schema-gated 4-stage innovation discovery and engineering blueprint pipeline built for **Google Antigravity `/swarm`** and **local Ollama models**.

Instead of passing free-form prose between agents, each stage emits a strict **Pydantic v2** JSON artifact validated by deterministic Python gates before the next stage unlocks:

1. **Stage 1 (`01_opportunities.json`):** Parallel user-group scouting for quantified operational bottlenecks.
2. **Stage 2 (`02_analyses.json`):** First-Principles decomposition, Cause-Effect Chain Analysis, Technical & Physical Contradiction framing, and ICE scoring with funnel pruning.
3. **Stage 3 (`03_recommendations.json`):** Ideal Final Result (IFR) synthesis and contradiction elimination via TRIZ separation principles and free existing resources.
4. **Stage 4 (`04_implementations.json` & `FINAL_REPORT.md`):** Phased engineering blueprints with bill of materials/stack, machine-verifiable gates, and quantitative kill-switch criteria.

---

## Repository Layout

innovation-swarm-engine/
├── GEMINI.md                     # Persistent Antigravity /swarm rules & Verifier gates
├── pyproject.toml                # Python 3.12+, Pydantic v2, mypy --strict, ruff & pytest config
├── README.md                     # Architecture & execution guide
├── schemas/
│   └── pipeline.py               # Strict Pydantic v2 contracts (ConfigDict extra="forbid", strict=True)
├── scripts/
│   ├── validate_stage.py         # Schema, cross-stage foreign-key & funnel pruning gatekeeper
│   ├── render_report.py          # Transitive Stage 4 validator + Markdown dossier compiler
│   └── run_local_ollama.py       # Standalone local runner using Ollama structured JSON outputs
├── prompts/
│   └── run_pipeline.md           # Parameterized /swarm prompt template
├── tests/                        # Pytest verification suite
└── artifacts/
    └── <run_id>/                 # Isolated per-run JSON artifacts & FINAL_REPORT.md

Installation & Static Verification

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install -e ".[dev]"

# Verify code quality, strict typing, and tests
ruff check schemas scripts tests
mypy --strict schemas scripts tests
pytest

Execution Mode A: Google Antigravity /swarm

1. Populate the run variables in prompts/run_pipeline.md (<run_id>, target domain, user groups, hard constraints, and funnel selection cap).

2. Submit the prompt in Antigravity with /swarm.

   • Coordinator & Verifier: Pinned to gemini-4-carbon to enforce First-Principles, Contradiction, and IFR gates defined in GEMINI.md.

   • Parallel Workers: Routed to gemini-4-argon inside isolated Git worktrees per user group (Stage 1) and per opportunity (Stage 2).

3. Each stage transition is gated by:

   python scripts/validate_stage.py --run-id <run_id> --stage <1|2|3|4>
   

4. Stage 4 concludes by compiling the final dossier:

   python scripts/render_report.py --run-id <run_id>
   

Execution Mode B: Standalone Local Ollama Runner

Run the exact same 4-stage pipeline, structured JSON schema repair loops, and adversarial critic checks locally against http://localhost:11434:

python scripts/run_local_ollama.py \
  --run-id 2026-11-13_marina_ops \
  --domain "Coastal Marina & Dock Logistics" \
  --user-groups "Small marina operators, recreational vessel owners, marine service technicians" \
  --constraints "CapEx under $5,000 per unit; zero cloud dependency required at dockside" \
  --opps-per-group 3 \
  --max-selected 3 \
  --worker-model granite4:latest \
  --critic-model nemotron-3:30b

CLI Options (run_local_ollama.py)

Flag	Default	Purpose
--run-id	(required)	Folder name under artifacts/<run-id>/ and artifact run_id key.
--domain	(required)	Target industry or technical problem space.
--user-groups	(required)	Comma-separated user segments scouted in Stage 1.
--constraints	Passive / low-CapEx bias	Hard engineering, budget, or operational boundaries.
--opps-per-group	3	Opportunities generated per user group in Stage 1.
--max-selected	3	Funnel pruning cap advancing from Stage 2 to Stages 3–4.
--worker-model	granite4:latest	Ollama model for Stages 1–3 generation.
--critic-model	nemotron-3:30b	Ollama model for Adversarial Critic loops & Stage 4 architecture.
--max-critic-loops	2	Maximum adversarial revision cycles per item before advancing.

Deterministic Gatekeeper (validate_stage.py)

Every stage artifact must pass three layers of verification before downstream stages execute:

• Schema Integrity: Enforces extra="forbid", strict=True, regex ID formats (OPP-001, ANL-001, REC-001, IMP-001), opposing physical contradiction states on the same control element, and exact ICE arithmetic (round((impact * confidence * ease) / 10.0, 2)).

• Cross-Stage Foreign-Key Traceability: Enforces strict relational lineage (IMP -> REC -> ANL -> OPP) and 1:1 completeness between Stage 1 opportunities and Stage 2 analyses.

• Funnel Pruning Gates: Enforces the --max-selected cap, prevents rank inversion (lower ICE scores cannot advance ahead of higher ICE scores unless --allow-rank-override is passed), and blocks pruned opportunities (selected_for_stage_3=False) from leaking into Stages 3 or 4.

---

### Quick Verification Check
Once you save `pyproject.toml` and `README.md` alongside `schemas/pipeline.py` and `scripts/*.py`, run:
```bash
pip install -e ".[dev]" && mypy --strict schemas scripts && ruff check schemas scripts

That confirms zero type or lint errors across the entire codebase before your first /swarm or Ollama run.

From <https://gemini.google.com/app/1a2e91195e6c2339>  Google Gemini (4 Hydrogen)