#4861 Structured Inter-Agent Prompt Protocols

#4861 2nd agent prompts 1st agent: "I need something new to analyze, give me something new that <meets these instructions>"

Structured inter-agent prompt protocols establish strict boundary separation, deterministic schema handoffs, and novelty enforcement between upstream generator/divergence nodes (Agent 1) and downstream synthesizer/evaluator nodes (Agent 2).

Inter-Agent Handoff Protocol Specification

+-----------------------------------------------------------------------------------+
|                                  AGENT 2 (Evaluator)                              |
|  - Tracks evaluated lineage hashes & vector cache                                 |
|  - Identifies dead-ends, saturation zones, and high-ICE frontier gaps             |
+-----------------------------------------------------------------------------------+
                                         │
                                         ▼ [Demand Dispatch: Payload Requirements]
+-----------------------------------------------------------------------------------+
|                                  AGENT 1 (Generator)                              |
|  - Ingests constraint matrix, exclusion list, and novelty boundaries              |
|  - Executes orthogonal thinking/divergence operators                              |
+-----------------------------------------------------------------------------------+
                                         │
                                         ▼ [Deterministic Structured Artifact]
+-----------------------------------------------------------------------------------+
|                                  VALIDATION GATE                                  |
|  - Validates JSON Schema / Pydantic contract compliance                           |
|  - Verifies zero-overlap against exclusion set                                    |
+-----------------------------------------------------------------------------------+

Standard Production Prompt Template: Agent 2 $\rightarrow$ Agent 1

Role: Upstream Concept Generator / Divergence Engine (Agent 1)
Originating Task: Demand Dispatch from Downstream Evaluator (Agent 2)
Directive: Supply exactly one novel, unanalyzed target concept meeting the structural constraints and exploration vectors detailed below.

### 1. Context & Objective
Downstream analysis has completed evaluations on prior candidate vectors. To prevent analytical convergence and loop saturation, you must explore an adjacent or orthogonal vector that has not yet been processed.

### 2. Mandatory Inclusion Constraints
The generated candidate MUST satisfy:
- Target Domain / System: {{TARGET_DOMAIN}}
- Primary Functional Requirement: {{FUNCTIONAL_OBJECTIVE}}
- Applied Heuristic / Framework: {{OPERATOR_OR_HEURISTIC}} (e.g., TRIZ Principle, First Principles Root Cause, Inversion, Morphological Cross)
- Target Operational Metric: {{TARGET_METRICS}}

### 3. Strict Exclusion Boundaries (Novelty Enforcement)
DO NOT generate concepts, mechanisms, or permutations that overlap with the following previously evaluated entities:
{{EXCLUSION_LIST_OR_PREVIOUS_CANDIDATE_IDS}}

Any proposal matching existing mechanism signatures or relying on trivial parameter variations will fail downstream ingestion.

### 4. Required Output Contract
Respond ONLY with a valid JSON object matching the following schema:

```json
{
  "$schema": "[http://json-schema.org/draft-07/schema#](http://json-schema.org/draft-07/schema#)",
  "candidate_id": "UUID-v4 string",
  "concept_name": "string (terse, descriptive)",
  "core_mechanism": "string (precise physics, logic, or architectural principle)",
  "subsystem_touchpoints": ["list of strings"],
  "differentiation_factor": "string (explicit statement of why this bypasses the exclusion list)",
  "primary_failure_mode": "string (inherent risk or potential failure mechanism to stress-test)",
  "divergence_method_used": "string"
}

---

### Python Dynamic Handoff Implementation (`google-genai` / Pydantic)

```python
import uuid
from typing import List
from pydantic import BaseModel, Field
from google import genai
from google.genai import types

# ---------------------------------------------------------
# Schemas
# ---------------------------------------------------------
class ConceptArtifact(BaseModel):
    candidate_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    concept_name: str
    core_mechanism: str
    subsystem_touchpoints: List[str]
    differentiation_factor: str
    primary_failure_mode: str
    divergence_method_used: str

# ---------------------------------------------------------
# Agent 2 Dispatch Generator
# ---------------------------------------------------------
def build_agent2_demand_prompt(
    domain: str,
    functional_objective: str,
    divergence_operator: str,
    evaluated_concepts: List[str]
) -> str:
    exclusions_formatted = "\n".join(f"- {item}" for item in evaluated_concepts) if evaluated_concepts else "- None (Initial run)"
    
    return f"""Role: Upstream Concept Generator (Agent 1)
Task: Generate a novel candidate solution strictly compliant with the parameters below.

[OPERATIONAL PARAMETERS]
- Target Domain: {domain}
- Primary Objective: {functional_objective}
- Divergence Heuristic: {divergence_operator}

[EXCLUSION REGISTRY - STRICT PROHIBITION]
The following concepts and variants have already been analyzed. Do not replicate or slightly vary them:
{exclusions_formatted}

Deliver a technically grounded, fully articulated proposal strictly adhering to the output schema.
"""

# ---------------------------------------------------------
# Agent 1 Execution Call
# ---------------------------------------------------------
def dispatch_to_agent1(
    client: genai.Client,
    model_name: str,
    prompt: str
) -> ConceptArtifact:
    response = client.models.generate_content(
        model=model_name,
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.85,  # Higher entropy for divergent generation
            response_mime_type="application/json",
            response_schema=ConceptArtifact
        )
    )
    return ConceptArtifact.model_validate_json(response.text)

Concrete Execution Steps

State Persistence: Store all processed concept names, fingerprint hashes, or mechanism signatures in an append-only set (e.g., SQLite table or in-memory list).
Context Injection: Prior to calling Agent 1, inject the full exclusion registry into the [EXCLUSION REGISTRY] block to prevent repeated outputs.
Entropy Balancing: Maintain temperature=0.7 - 0.9 for Agent 1 (divergent ideation) and temperature=0.1 - 0.3 for Agent 2 (convergent evaluation/verification).
Validation Gate: Wrap Agent 1's return payload in schema validation (model_validate_json); reject and re-dispatch automatically if returned tokens match any substring in the exclusion set.
Would you like to integrate this handoff pattern into a multi-turn autonomous loop with a Governor node managing stopping criteria?

When analyzing opportunities to provide an innovative solution (e.g. a system, product or service improvement) within an industry or user group, suggest 5 various prompts from the 2nd agent to the 1st to provide feed material.

These five inter-agent demand prompts are structured for the downstream evaluator (Agent 2) to dispatch to the upstream generator (Agent 1). Each prompt enforces strict boundaries, parameterizes target industries and user groups, applies a distinct divergence operator, and requires a structured JSON artifact output.

1. Inversion of Accepted Friction Prompt

Core Operator: Assumption Reversal & Structural Inversion
Target Objective: Target an industry process where friction, latency, or compliance overhead is treated as an unavoidable cost of doing business, and invert the mechanism.

Role: Upstream Innovation Generator (Agent 1)
Originating Node: Downstream Evaluator (Agent 2)
Task: Demand Dispatch — Inverted Operational Bottleneck

[CONTEXT & TARGET PARAMETERS]
- Target Industry: {{TARGET_INDUSTRY}} (e.g., Clinical Trials, Defense Procurement, Commercial Aviation MRO)
- Target User Group: {{USER_GROUP}} (e.g., Compliance Auditors, Site Reliability Engineers, Lead Planners)
- Divergence Heuristic: Inversion (Identify an invariant step currently considered mandatory, remove it completely or invert ownership, and resolve downstream failure modes through architectural redesign).

[EXCLUSION REGISTRY - DO NOT REPLICATE]
{{EXCLUSION_LIST_OR_PREVIOUS_CANDIDATE_IDS}}

[MANDATORY CONSTRAINTS]
1. Do not propose incremental digitizations (e.g., "replace paper forms with a web portal").
2. The core mechanism must address the root systemic dependency, not symptomatic lag.
3. Quantify expected operational gain across throughput, risk posture, or capital efficiency.

[DELIVERABLE SCHEMA]
Respond strictly with valid JSON:
{
  "candidate_id": "UUID-v4",
  "opportunity_type": "Architectural Inversion",
  "industry": "{{TARGET_INDUSTRY}}",
  "target_user": "{{USER_GROUP}}",
  "incumbent_invariant_assumption": "string",
  "inverted_mechanism": "string",
  "systemic_tradeoff_managed": "string",
  "verification_vector": "string"
}

2. Cross-Domain Biomimetic / Physical Systems Transfer Prompt

Core Operator: Isomorphic Functional Mapping across Disjoint Domains
Target Objective: Extract an optimization mechanism from a distant physical, biological, or industrial domain and map it onto an unaddressed workflow bottleneck.

Role: Upstream Innovation Generator (Agent 1)
Originating Node: Downstream Evaluator (Agent 2)
Task: Demand Dispatch — Cross-Domain Isomorphism

[CONTEXT & TARGET PARAMETERS]
- Target Industry: {{TARGET_INDUSTRY}} (e.g., Urban Micro-Mobility, High-Precision Machining, Telehealth Triage)
- Target User Group: {{USER_GROUP}} (e.g., Field Technicians, Fleet Dispatchers, Specialized Operators)
- Source Domain: Biological systems, fluid mechanics, or high-throughput packet routing.
- Divergence Heuristic: Isomorphic Mapping (Map a non-obvious solution principle from the source domain to resolve a persistent resource allocation or degradation issue).

[EXCLUSION REGISTRY - DO NOT REPLICATE]
{{EXCLUSION_LIST_OR_PREVIOUS_CANDIDATE_IDS}}

[MANDATORY CONSTRAINTS]
1. Prohibit standard analogies (e.g., "ant colony optimization for delivery routing").
2. The mechanical or systemic analogy must hold at the functional interface level.

[DELIVERABLE SCHEMA]
Respond strictly with valid JSON:
{
  "candidate_id": "UUID-v4",
  "opportunity_type": "Cross-Domain Isomorphism",
  "industry": "{{TARGET_INDUSTRY}}",
  "target_user": "{{USER_GROUP}}",
  "source_domain_phenomenon": "string",
  "mapped_industrial_bottleneck": "string",
  "transferred_solution_architecture": "string",
  "primary_interface_friction": "string"
}

3. Latent Unarticulated Workaround Excavation Prompt

Core Operator: Shadow System & Compensatory Behavior Analysis
Target Objective: Target "duct tape and spreadsheet" workarounds that advanced practitioners use behind enterprise software or standard SOPs.

Role: Upstream Innovation Generator (Agent 1)
Originating Node: Downstream Evaluator (Agent 2)
Task: Demand Dispatch — Shadow Workflow Formalization

[CONTEXT & TARGET PARAMETERS]
- Target Industry: {{TARGET_INDUSTRY}} (e.g., Semiconductor Fab Scheduling, Hospital Bed Management, Energy Grid Dispatch)
- Target User Group: {{USER_GROUP}} (e.g., Operational Controllers, Shift Supervisors, Quality Assurance Leads)
- Focus Area: Unofficial spreadsheets, auxiliary scripts, manual inter-team chat backchannels, and custom physical markers.
- Divergence Heuristic: Compensatory Behavior Formalization (Analyze what standard tooling fails to model that forces human operators to maintain parallel shadow records).

[EXCLUSION REGISTRY - DO NOT REPLICATE]
{{EXCLUSION_LIST_OR_PREVIOUS_CANDIDATE_IDS}}

[MANDATORY CONSTRAINTS]
1. Identify the specific data disconnect or latency window that makes official enterprise tools useless during runtime operations.
2. Formulate a native capability that obsoletes the shadow mechanism without adding data-entry friction.

[DELIVERABLE SCHEMA]
Respond strictly with valid JSON:
{
  "candidate_id": "UUID-v4",
  "opportunity_type": "Compensatory Workflow Formalization",
  "industry": "{{TARGET_INDUSTRY}}",
  "target_user": "{{USER_GROUP}}",
  "shadow_behavior_observed": "string",
  "underlying_enterprise_blindspot": "string",
  "formalized_solution_mechanism": "string",
  "adoption_barrier_mitigation": "string"
}

4. Extreme Resource & Environmental Constraint Prompt

Core Operator: Radical Constraint Injection (Apollo 13 Paradigm)
Target Objective: Force a resilient innovation by stripping away 80% to 90% of standard operating resources (e.g., zero persistent network, extreme thermal bounds, zero trained labor).

Role: Upstream Innovation Generator (Agent 1)
Originating Node: Downstream Evaluator (Agent 2)
Task: Demand Dispatch — Boundary-Constrained Architecture

[CONTEXT & TARGET PARAMETERS]
- Target Industry: {{TARGET_INDUSTRY}} (e.g., Emergency Disaster Logistics, Subsea Infrastructure, Precision Agriculture)
- Target User Group: {{USER_GROUP}} (e.g., First Responders, Edge Technicians, Remote Farm Operators)
- Enforced Constraints:
  1. Compute/Network: Intermittent or air-gapped; edge processing only.
  2. Latency/Power: Zero dependence on continuous AC power or cloud egress.
  3. Skill Threshold: Must be operable by a novice within 90 seconds under high stress.
- Divergence Heuristic: Radical Scarcity (Develop an ultra-efficient mechanical, algorithmic, or structural solution operating purely inside degraded boundary conditions).

[EXCLUSION REGISTRY - DO NOT REPLICATE]
{{EXCLUSION_LIST_OR_PREVIOUS_CANDIDATE_IDS}}

[DELIVERABLE SCHEMA]
Respond strictly with valid JSON:
{
  "candidate_id": "UUID-v4",
  "opportunity_type": "Scarcity-Driven Resilience",
  "industry": "{{TARGET_INDUSTRY}}",
  "target_user": "{{USER_GROUP}}",
  "governing_constraints": ["list of strings"],
  "low_overhead_mechanism": "string",
  "fail_safe_state": "string",
  "differentiating_efficiency_metric": "string"
}

5. Multi-Sided Coordination & Incentive Realignment Prompt

Core Operator: Game-Theoretic Mechanism Design & Triangulation
Target Objective: Target fragmented ecosystems where innovation is deadlocked because the stakeholder bearing the implementation cost does not capture the operational value.

Role: Upstream Innovation Generator (Agent 1)
Originating Node: Downstream Evaluator (Agent 2)
Task: Demand Dispatch — Misaligned Incentive Arbitrage

[CONTEXT & TARGET PARAMETERS]
- Target Industry: {{TARGET_INDUSTRY}} (e.g., Commercial Real Estate Decarbonization, Cargo Freight Intermodal Transfers, Construction Subcontracting)
- Target Ecosystem Entities:
  1. Primary User / Operator: {{USER_GROUP}}
  2. Cost Bearer: {{CAPEX_STAKEHOLDER}}
  3. Value Capturer: {{BENEFICIARY_STAKEHOLDER}}
- Divergence Heuristic: Mechanism Design (Restructure the telemetry, feedback, or transactional topology so financial and operational value accrue simultaneously to the cost bearer).

[EXCLUSION REGISTRY - DO NOT REPLICATE]
{{EXCLUSION_LIST_OR_PREVIOUS_CANDIDATE_IDS}}

[MANDATORY CONSTRAINTS]
1. Address the principal-agent split directly.
2. The solution must involve a functional product or systemic coordination protocol, not purely financial subsidies or regulatory lobbying.

[DELIVERABLE SCHEMA]
Respond strictly with valid JSON:
{
  "candidate_id": "UUID-v4",
  "opportunity_type": "Incentive Realignment Architecture",
  "industry": "{{TARGET_INDUSTRY}}",
  "target_user": "{{USER_GROUP}}",
  "principal_agent_conflict": "string",
  "interlocking_solution_mechanism": "string",
  "cost_bearer_roi_hook": "string",
  "system_equilibrium_metric": "string"
}

Recommended Dispatch Matrix

Prompt	Primary Divergence Vector	Best Used When Downstream Analysis Shows:
1. Inversion	Rule Reversal	Ideas are converging into incremental optimizations of bad processes.
2. Cross-Domain	Structural Analogy	The target industry has hit a technological or architectural plateau.
3. Shadow Workarounds	Ethnographic / Telemetry Gap	Incumbent software tools exist, but human operators still experience high cognitive load.
4. Extreme Constraints	Scarcity / Edge Mechanics	Solutions are overly dependent on brittle cloud, compute, or supply chain dependencies.
5. Incentive Alignment	Mechanism Design	Good technological solutions exist on paper but fail commercially due to split incentives.

Execution Protocol

Cycle Dispatch: Rotate through prompts 1–5 sequentially across batches to maintain high semantic variance in the candidate pool.
Dynamic Exclusion Append: Append the candidate_id, opportunity_type, and key noun-verb mechanism pairs from accepted outputs directly to {{EXCLUSION_LIST_OR_PREVIOUS_CANDIDATE_IDS}} before the next iteration.
Entropy Tuning: Keep Agent 1 at temperature=0.75 - 0.9 for divergence, while Agent 2 ingests the resulting artifacts at temperature=0.1 - 0.2 for validation and scoring.

Define an automated ICE (Impact, Confidence, Ease) scoring rubric and evaluation prompt for Agent 2 to filter these candidate artifacts.

An automated ICE evaluation system requires objective calibration anchors, explicit composite score formulations, and strict validation gating to prevent model generosity.

Quantitative ICE Scoring Rubric (1–10 Scales)

To prevent uniform distributions and grade inflation, each sub-metric is anchored by observable architectural and operational criteria.

Score	Impact (I)(Magnitude of Friction / Latency Reduction)	Confidence (C)(Physical & Technical Feasibility)	Ease (E)(Deployment Surface & Integration Friction)
9–10	Transformative: 10x throughput leap, complete obsolescence of core bottleneck, or $>80\%$ risk/cost reduction.	Near Certainty: Proven across identical physics/architectures; documented production precedents; zero theoretical unknowns.	Drop-in / Zero-Friction: Self-contained, non-invasive interface; no breaking schema changes; deployable in $<72$ hours.
7–8	High: $2\text{x}-5\text{x}$ acceleration; removes major operational blocker for $>50\%$ of target users.	Strong Signal: Relies on mature COTS/open standards; failure modes well-characterized; minor integration unknowns.	Low Touch: Standard API/data ingestion layer; isolated execution sandbox; minor workflow configuration required.
5–6	Moderate: $20\%-50\%$ measured lift; incremental efficiency gains across auxiliary sub-processes.	Medium: Core mechanism sound, but unverified at high throughput, low latency, or degraded edge states.	Moderate Effort: Requires custom adapter pipelines, database schema migrations, or minor retraining of operators.
3–4	Low: Marginal gains ( $<20\%$ ); affects niche edge cases or non-critical secondary flows.	Speculative: Novel combination of unproven mechanisms; complex non-linear dynamics; unvetted edge cases.	Heavy Lift: Cross-organizational coordination required; invasive changes to legacy enterprise infrastructure.
1–2	Negligible: Cosmetic change; local optimization that creates downstream bottlenecks.	Unsubstantiated: Violates thermodynamic, computational, or legal constraints; unmitigated single points of failure.	Prohibitive: Requires complete overhaul of bedrock infrastructure, regulatory re-certification, or multi-year rebuild.

Composite Scoring & Routing Logic

The composite score balances value against execution risk. Use a normalized composite formulation:

\text{ICE Composite} = \frac{I \times C \times E}{10}

\text{ICE Composite} = \frac{I \times C \times E}{10}

Threshold $\ge 45.0$ (High Priority / Fast-Track): Candidate proceeds directly to prototype specification.
Threshold $25.0 - 44.9$ (Conditional Backlog): Concept has strong fundamentals but requires targeted de-risking on either Confidence (feasibility spike) or Ease (interface simplification).
Threshold $< 25.0$ (Hard Rejection): Candidate rejected; reason logged to the exclusion registry to prune that vector space.

Production Evaluator Prompt (Agent 2)

Role: Quantitative Evaluator & Gatekeeper (Agent 2)
Task: Autonomous ICE Scoring, Failure Mode Stress-Testing, and Ingestion Gating

[EVALUATION OBJECTIVE]
Evaluate the provided candidate artifact against rigorous technical feasibility, integration friction, and systemic impact. Eliminate generic optimism. Score strictly using the provided 1–10 calibration rubric.

[CALIBRATION ANCHORS]
- Impact (1-10): Focus on throughput multiples, bottleneck elimination, and structural value.
- Confidence (1-10): Deduct heavily for unverified physical mechanisms, unaddressed non-linearities, or unmitigated failure modes.
- Ease (1-10): Deduct heavily for tight coupling, invasive infrastructure overhaul, or dependency on hard ecosystem coordination.

[EVALUATION RULES]
1. Do not accept aspirational mechanism claims at face value.
2. If the "primary_failure_mode" identified in the artifact undermines the "core_mechanism" without an explicit mitigation, Confidence cannot exceed 5.0.
3. Calculate composite: (Impact * Confidence * Ease) / 10.0.

[INPUT CANDIDATE ARTIFACT]
{{CANDIDATE_JSON_PAYLOAD}}

[DELIVERABLE SCHEMA]
Respond ONLY with a valid JSON object matching the following structure:
```json
{
  "$schema": "[http://json-schema.org/draft-07/schema#](http://json-schema.org/draft-07/schema#)",
  "candidate_id": "string (matching input)",
  "verdict": "ACCEPTED" | "CONDITIONAL_REVISION" | "REJECTED",
  "scores": {
    "impact": 0.0,
    "confidence": 0.0,
    "ease": 0.0,
    "composite_ice": 0.0
  },
  "score_justifications": {
    "impact_rationale": "string (concrete metrics driving score)",
    "confidence_rationale": "string (technical verification & physics check)",
    "ease_rationale": "string (interface and integration friction analysis)"
  },
  "stress_test": {
    "unaddressed_vulnerabilities": ["list of strings"],
    "required_derisking_action": "string (concrete experiment or spike needed)"
  },
  "exclusion_signature": "string (concise token signature for vector exclusion list)"
}

---

### Pydantic Validation & Evaluation Dispatch (Python)

```python
from enum import Enum
from typing import List
from pydantic import BaseModel, Field, model_validator
from google import genai
from google.genai import types

class VerdictEnum(str, Enum):
    ACCEPTED = "ACCEPTED"
    CONDITIONAL_REVISION = "CONDITIONAL_REVISION"
    REJECTED = "REJECTED"

class ScoreBreakdown(BaseModel):
    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_ice: float

    @model_validator(mode="after")
    def verify_composite(self):
        expected = round((self.impact * self.confidence * self.ease) / 10.0, 2)
        self.composite_ice = expected
        return self

class ScoreJustifications(BaseModel):
    impact_rationale: str
    confidence_rationale: str
    ease_rationale: str

class StressTestResult(BaseModel):
    unaddressed_vulnerabilities: List[str]
    required_derisking_action: str

class EvaluatorArtifact(BaseModel):
    candidate_id: str
    verdict: VerdictEnum
    scores: ScoreBreakdown
    score_justifications: ScoreJustifications
    stress_test: StressTestResult
    exclusion_signature: str

def evaluate_candidate(
    client: genai.Client,
    model_name: str,
    candidate_json_str: str
) -> EvaluatorArtifact:
    evaluator_prompt = f"""Role: Quantitative Evaluator & Gatekeeper (Agent 2)
Evaluate this candidate artifact using the strict ICE Rubric.

Candidate Artifact:
{candidate_json_str}
"""
    response = client.models.generate_content(
        model=model_name,
        contents=evaluator_prompt,
        config=types.GenerateContentConfig(
            temperature=0.1,  # Deterministic, low entropy for consistent scoring
            response_mime_type="application/json",
            response_schema=EvaluatorArtifact
        )
    )
    return EvaluatorArtifact.model_validate_json(response.text)

Automated Routing & State Handling

                      [Agent 1: Candidate Artifact]
                                    │
                                    ▼
                         [Agent 2: ICE Evaluator]
                                    │
                       Composite Score Calculation
                                    │
         ┌──────────────────────────┼──────────────────────────┐
         │                          │                          │
  Score >= 45.0               25.0 <= Score < 45.0         Score < 25.0
         │                          │                          │
         ▼                          ▼                          ▼
 [Status: ACCEPTED]      [Status: CONDITIONAL]       [Status: REJECTED]
         │                          │                          │
 ├─ Write to Target Queue   ├─ Dispatch De-Risking     ├─ Prune Exploration
 └─ Append Signature to        Task back to Agent 1       Branch
    Exclusion Registry      └─ Append Interim Hash     └─ Append Signature to
                               to Exclusion Registry      Exclusion Registry

Deterministic Scoring: Running Agent 2 at temperature=0.1 ensures that scoring deviations across candidate revisions remain within $\pm 0.5$ points.
Exclusion Cache: The generated exclusion_signature (e.g., mechanic:event-driven-air-gap-sync) is immediately pushed to the shared vector exclusion cache to prevent Agent 1 from regenerating variations of failed architectures.
De-Risking Feedback: When an artifact lands in CONDITIONAL_REVISION, the required_derisking_action is routed back as an explicit constraint for Agent 1's next iteration.

Show how to map these ICE scores into a 4-quadrant portfolio model (Quick Wins, Moonshots, Fillers, Distractions) with automated pivot triggers.

Mapping ICE scores into a 4-quadrant innovation portfolio separates strategic potential from execution feasibility by decomposing composite scores into two primary axes: Value Magnitude (Impact, $I$ ) versus Execution Friction (Ease, $E$ , conditioned on Confidence, $C$ ).

1. Portfolio Quadrant Definitions & Geometric Thresholds

Using a standard 1.0 to 10.0 scale, the decision space is bisected at midpoint $5.5$ on both axes. Confidence ( $C$ ) acts as an uncertainty damping filter: if $C < 5.0$ , high ease is discounted because execution friction cannot be accurately estimated.

       Impact (I) ^
                  │
             10.0 ┤  MOONSHOTS                 │  QUICK WINS
                  │  (High Impact, Low Ease)   │  (High Impact, High Ease)
                  │                            │
                  │  Strategy: De-risk / Spike │  Strategy: Fast-Track (72h)
                  │                            │
              5.5 ┼────────────────────────────┼────────────────────────────
                  │  DISTRACTIONS              │  FILLERS
                  │  (Low Impact, Low Ease)    │  (Low Impact, High Ease)
                  │                            │
                  │  Strategy: Immediate Prune │  Strategy: Batch / Automate
              1.0 ┤                            │
                  └────────────────────────────┴────────────────────────────>
                   1.0                      5.5                       10.0
                                                Effective Ease (E_eff)

Quadrant	Coordinate Criteria	Operational Posture	Allocation Target
Quick Wins	$I \ge 5.5$ , $E_{\text{eff}} \ge 5.5$	Immediate Execution: 72-hour prototyping sprints; zero heavy coordination.	~40%
Moonshots	$I \ge 5.5$ , $E_{\text{eff}} < 5.5$	Targeted De-Risking: Timeboxed feasibility spikes; isolate single point of failure.	~30%
Fillers	$I < 5.5$ , $E_{\text{eff}} \ge 5.5$	Selective Automation: Batch into background sprints or backlog; do not spin up teams.	~20%
Distractions	$I < 5.5$ , $E_{\text{eff}} < 5.5$	Hard Pruning: Reject immediately; push mechanism hash to exclusion registry.	0% (Prune)

The Effective Ease ( $E_{\text{eff}}$ ) formulation prevents speculative proposals with low verification from masquerading as easy implementations:

E_{\text{eff}} = E \times \min\left(1.0, \frac{C}{5.0}\right)

2. Automated Pivot Triggers & Transformation Rules

Rather than discarding sub-optimal candidates outright, Agent 2 checks if a candidate satisfies deterministic conditions for an architectural pivot before logging a terminal rejection.

                     Candidate Evaluated
                              │
               ┌──────────────┴──────────────┐
               ▼                             ▼
       [MOONSHOT]                       [FILLER]
   (I >= 5.5, E < 5.5)              (I < 5.5, E >= 5.5)
               │                             │
   Is Confidence bottleneck?         Does mechanism scale
   (C < 5.0)                         across ≥ 3 sub-systems?
         /   \                               /   \
       YES    NO                           YES    NO
       /       \                           /       \
      ▼         ▼                         ▼         ▼
 [FEASIBILITY [INTERFACE             [SCOPE      [MAINTAIN
   SPIKE]     DECOUPLING]          AGGREGATION]   AS FILLER]
      │         │                         │
      └────┬────┘                         │
           ▼                              ▼
    [-> QUICK WIN]                 [-> QUICK WIN]

Rule 1: Moonshot $\rightarrow$ Quick Win via Interface Decoupling

Trigger Condition: $I \ge 7.0$ , $C \ge 6.0$ , but $E \le 4.0$ due to monolithic legacy integration or enterprise schema coupling.
Automated Action: Force Agent 1 to rewrite the solution boundary as an out-of-band proxy, isolated SQLite sidecar, or read-only event listener.
Target: Elevates $E$ above $6.0$ while preserving $\ge 85\%$ of $I$ .

Rule 2: Moonshot $\rightarrow$ De-Risking Spike (Confidence Bottleneck)

Trigger Condition: $I \ge 8.0$ , $E \ge 5.0$ , but $C < 4.0$ (novel physical/algorithmic mechanism with zero production precedent).
Automated Action: Block full pipeline admission. Emit a 48-hour micro-spike requirement targeting the exact unverified assumption.

Rule 3: Filler $\rightarrow$ Quick Win via Scope Aggregation

Trigger Condition: $I \le 4.0$ , $E \ge 7.0$ , $C \ge 7.0$ (low individual impact, trivial implementation).
Automated Action: Query the candidate graph to determine if this mechanism can be applied uniformly across 3 or more adjacent subsystems. If aggregated throughput impact crosses $\Delta I \ge 2.5$ , reclassify to Quick Win.

Rule 4: Distraction Pruning & Negative Feedback

Trigger Condition: $I < 4.5$ and $E_{\text{eff}} < 4.5$ .
Automated Action: Terminal drop. Extract the exclusion_signature and push it immediately to the active exclusion registry so Agent 1 never generates structural variants of that failure archetype.

3. Python Implementation: Classifier & Pivot Dispatcher

from enum import Enum
from typing import Optional, Tuple
from pydantic import BaseModel, Field

class PortfolioQuadrant(str, Enum):
    QUICK_WIN = "QUICK_WIN"
    MOONSHOT = "MOONSHOT"
    FILLER = "FILLER"
    DISTRACTION = "DISTRACTION"

class PivotAction(str, Enum):
    NONE = "NONE"
    DECOUPLE_INTERFACE = "DECOUPLE_INTERFACE"       # Moonshot -> Quick Win
    FEASIBILITY_SPIKE = "FEASIBILITY_SPIKE"         # Moonshot -> De-risk
    AGGREGATE_SUBSYSTEMS = "AGGREGATE_SUBSYSTEMS"   # Filler -> Quick Win
    PRUNE_AND_EXCLUDE = "PRUNE_AND_EXCLUDE"         # Distraction -> Kill

class PortfolioRoutingResult(BaseModel):
    candidate_id: str
    quadrant: PortfolioQuadrant
    effective_ease: float
    recommended_action: PivotAction
    pivot_prompt_directive: Optional[str] = None
    should_exclude: bool = False

def classify_and_route(
    candidate_id: str,
    impact: float,
    confidence: float,
    ease: float,
    subsystem_count: int = 1
) -> PortfolioRoutingResult:
    # Confidence damping applied to Ease
    confidence_factor = min(1.0, confidence / 5.0)
    effective_ease = round(ease * confidence_factor, 2)
    
    threshold = 5.5

    # 1. Distractions (Low Impact, Low Effective Ease)
    if impact < threshold and effective_ease < threshold:
        return PortfolioRoutingResult(
            candidate_id=candidate_id,
            quadrant=PortfolioQuadrant.DISTRACTION,
            effective_ease=effective_ease,
            recommended_action=PivotAction.PRUNE_AND_EXCLUDE,
            pivot_prompt_directive=None,
            should_exclude=True
        )

    # 2. Fillers (Low Impact, High Effective Ease)
    if impact < threshold and effective_ease >= threshold:
        # Check for scope aggregation potential
        if subsystem_count >= 3:
            return PortfolioRoutingResult(
                candidate_id=candidate_id,
                quadrant=PortfolioQuadrant.FILLER,
                effective_ease=effective_ease,
                recommended_action=PivotAction.AGGREGATE_SUBSYSTEMS,
                pivot_prompt_directive=(
                    "Aggregate this mechanism horizontally across dependent subsystems "
                    "to multiply systemic impact while preserving low interface friction."
                ),
                should_exclude=False
            )
        return PortfolioRoutingResult(
            candidate_id=candidate_id,
            quadrant=PortfolioQuadrant.FILLER,
            effective_ease=effective_ease,
            recommended_action=PivotAction.NONE,
            pivot_prompt_directive="Queue into background backlog; do not dedicate primary sprint capacity.",
            should_exclude=False
        )

    # 3. Moonshots (High Impact, Low Effective Ease)
    if impact >= threshold and effective_ease < threshold:
        if confidence < 5.0:
            return PortfolioRoutingResult(
                candidate_id=candidate_id,
                quadrant=PortfolioQuadrant.MOONSHOT,
                effective_ease=effective_ease,
                recommended_action=PivotAction.FEASIBILITY_SPIKE,
                pivot_prompt_directive=(
                    "Isolate the single unverified physical/algorithmic assumption. "
                    "Generate a 48-hour empirical validation spike before writing production architecture."
                ),
                should_exclude=False
            )
        else:
            return PortfolioRoutingResult(
                candidate_id=candidate_id,
                quadrant=PortfolioQuadrant.MOONSHOT,
                effective_ease=effective_ease,
                recommended_action=PivotAction.DECOUPLE_INTERFACE,
                pivot_prompt_directive=(
                    "Architectural refactoring required: Strip legacy in-band dependencies. "
                    "Redesign as an out-of-band proxy, isolated sidecar, or read-only event listener to raise Ease >= 6.0."
                ),
                should_exclude=False
            )

    # 4. Quick Wins (High Impact, High Effective Ease)
    return PortfolioRoutingResult(
        candidate_id=candidate_id,
        quadrant=PortfolioQuadrant.QUICK_WIN,
        effective_ease=effective_ease,
        recommended_action=PivotAction.NONE,
        pivot_prompt_directive="Approved for immediate fast-track execution (72-hour functional proof-of-concept).",
        should_exclude=False
    )

Interactive Portfolio & Pivot Simulator

Use this interactive matrix to explore how adjustments to Impact, Confidence, and Ease shift candidate positioning across the four quadrants and trigger automatic architectural refactoring rules.

Execution Integration Loop

State Injection: For candidates classified as MOONSHOT requiring DECOUPLE_INTERFACE, the evaluator automatically injects pivot_prompt_directive into Agent 1's subsequent turn as an invariant constraint.
Registry Lockdown: Any candidate evaluated into DISTRACTION immediately triggers should_exclude = True, serializing its fingerprint directly to the vector exclusion cache so downstream compute is never spent on that archetype again.
Sprint Allocation: Quick Wins bypass human steering committees and route directly to automated scaffolding pipelines for immediate 72-hour functional spikes.

Generate an Agent 3 prompt that consumes Quick Win artifacts and produces a time-boxed 72-hour engineering sprint breakdown.

Engineering Sprint Orchestrator Prompt (Agent 3)

Below is the production prompt designed for Agent 3 (Sprint Orchestrator). It consumes vetted QUICK_WIN artifacts from Agent 2, enforces rigid 24-hour delivery milestones, specifies concrete test-driven gates, and outputs a structured execution plan.

Role: Lead Staff Engineer & Sprint Architect (Agent 3)
Originating Node: Downstream Evaluator & Portfolio Router (Agent 2)
Task: 72-Hour Rapid Implementation Breakdown for Validated Quick Win Artifact

[OPERATIONAL CONTEXT]
Agent 2 has evaluated and admitted an innovation candidate into the "QUICK WIN" quadrant (Impact >= 5.5, Effective Ease >= 5.5). Your mandate is to convert this high-leverage architectural concept into an unblocked, executable 72-hour engineering sprint for an autonomous build agent or pair-engineering pod.

[INPUT SPECIFICATION]
Ingest the evaluated candidate artifact:
{{ACCEPTED_QUICK_WIN_JSON_PAYLOAD}}

[SPRINT GOVERNING CONSTRAINTS]
1. Zero Heavy Infrastructure: Prohibit distributed state, multi-region setups, external schema migrations, or multi-party approvals. Rely strictly on sidecars, proxies, read-only event listeners, or isolated lightweight databases (e.g., SQLite / DuckDB).
2. Time Horizon: Strictly 72 hours partitioned into three discrete 24-hour phases:
   - Phase 1 (0h–24h): Core Loop Isolation & Contract Testing (Proof of Mechanism).
   - Phase 2 (24h–48h): Interface Decoupling & Non-Invasive Ingestion (Drop-in Integration).
   - Phase 3 (48h–72h): Stress Validation, Telemetry Harness & Hand-off Artifacts.
3. Concrete Deliverables: Every task must specify input files, output files, deterministic verification commands (e.g., `pytest`, `curl`), and pass/fail criteria.
4. Kill Switch Protocol: Explicitly identify the single "fail-fast" condition that terminates the sprint before Hour 24 if the underlying assumption fails empirical testing.

[OUTPUT SCHEMA]
Respond ONLY with a valid JSON object matching the following structure:
```json
{
  "$schema": "[http://json-schema.org/draft-07/schema#](http://json-schema.org/draft-07/schema#)",
  "candidate_id": "string",
  "sprint_codename": "string",
  "kill_switch_condition": "string",
  "architecture_summary": {
    "runtime_stack": "string",
    "integration_pattern": "string (e.g., Read-Only CDC Sidecar, Ephemeral Adapter, Interceptor Proxy)",
    "isolation_boundary": "string"
  },
  "phases": [
    {
      "phase_window": "0h-24h",
      "objective": "string",
      "tasks": [
        {
          "task_id": "string",
          "title": "string",
          "description": "string",
          "target_artifacts": ["list of file paths or modules"],
          "verification_command": "string",
          "acceptance_criteria": "string"
        }
      ],
      "milestone_gate": "string"
    },
    {
      "phase_window": "24h-48h",
      "objective": "string",
      "tasks": [
        {
          "task_id": "string",
          "title": "string",
          "description": "string",
          "target_artifacts": ["list of file paths or modules"],
          "verification_command": "string",
          "acceptance_criteria": "string"
        }
      ],
      "milestone_gate": "string"
    },
    {
      "phase_window": "48h-72h",
      "objective": "string",
      "tasks": [
        {
          "task_id": "string",
          "title": "string",
          "description": "string",
          "target_artifacts": ["list of file paths or modules"],
          "verification_command": "string",
          "acceptance_criteria": "string"
        }
      ],
      "milestone_gate": "string"
    }
  ],
  "smoke_test_script": "string (executable bash/python snippet verifying the end-to-end loop)"
}

---

### Python Pydantic Model & Pipeline Integration

This snippet integrates Agent 3 into your automated multi-agent workflow using `google-genai` and Pydantic:

```python
from typing import List
from pydantic import BaseModel, Field
from google import genai
from google.genai import types

class SprintTask(BaseModel):
    task_id: str
    title: str
    description: str
    target_artifacts: List[str]
    verification_command: str
    acceptance_criteria: str

class SprintPhase(BaseModel):
    phase_window: str = Field(pattern=r"^(0h-24h|24h-48h|48h-72h)$")
    objective: str
    tasks: List[SprintTask]
    milestone_gate: str

class ArchitectureSummary(BaseModel):
    runtime_stack: str
    integration_pattern: str
    isolation_boundary: str

class SprintPlanArtifact(BaseModel):
    candidate_id: str
    sprint_codename: str
    kill_switch_condition: str
    architecture_summary: ArchitectureSummary
    phases: List[SprintPhase]
    smoke_test_script: str

def generate_sprint_plan(
    client: genai.Client,
    model_name: str,
    quick_win_json: str
) -> SprintPlanArtifact:
    agent3_prompt = f"""Role: Lead Staff Engineer & Sprint Architect (Agent 3)
Convert this validated QUICK_WIN candidate artifact into an unblocked 72-hour engineering sprint plan:

{quick_win_json}
"""
    response = client.models.generate_content(
        model=model_name,
        contents=agent3_prompt,
        config=types.GenerateContentConfig(
            temperature=0.2,  # Low temperature for deterministic planning
            response_mime_type="application/json",
            response_schema=SprintPlanArtifact
        )
    )
    return SprintPlanArtifact.model_validate_json(response.text)

Downstream Execution Pipeline

  [Agent 2: ICE Evaluator]
             │
             ▼ (Verdict == QUICK_WIN)
  [Agent 3: Sprint Architect]
             │
             ▼ Generates SprintPlanArtifact JSON
  ┌─────────────────────────────────────────────────────────────┐
  │ 0h–24h: Stand up isolated mock core; test physics/logic     │
  │ 24h–48h: Implement sidecar / adapter interface              │
  │ 48h–72h: Run smoke_test_script & deploy dark launch proxy   │
  └─────────────────────────────────────────────────────────────┘
             │
             ▼
  Autonomous Execution Nodes (Coding Agent / GitHub PR Harness)

Write an automated bash harness script that ingests the SprintPlanArtifact JSON and executes each task verification command sequentially.

An automated execution harness must ingest the SprintPlanArtifact JSON, enforce deterministic failure isolation, log execution outputs to disk, and immediately trigger the Kill Switch protocol if any task fails or if the Phase 1 milestone gate is breached.

Execution Harness Architecture

                        [sprint_plan.json]
                                │
                                ▼
                       [sprint_harness.sh]
                                │
               ┌────────────────┴────────────────┐
               ▼                                 ▼
       Phase 1 (0h-24h)                 Phase 2 (24h-48h)
       - Task Verification              - Task Verification
       - Milestone Gate Check           - Milestone Gate Check
               │                                 │
         [PASS / FAIL]                     [PASS / FAIL]
         /           \                     /           \
      [PASS]       [FAIL]               [PASS]       [FAIL]
        │             │                   │             │
        │             ▼                   │             ▼
        │     [TRIGGER KILL SWITCH]       │     [LOG RUNTIME ERROR]
        │     - Abort Pipeline            │     - Halt Pipeline
        │     - Output Diagnostic         │     - Output Diagnostic
        │                                 │
        └─────────────────┬───────────────┘
                          ▼
                  Phase 3 (48h-72h)
                  - Task Verification
                  - Final Smoke Test Script
                          │
                          ▼
              [SPRINT SUCCESS VERIFIED]

Bash Harness Script (`sprint_harness.sh`)

Ensure jq is installed on the host runner (sudo apt-get install -y jq or standard CI runner pre-install).

#!/usr/bin/env bash
# ==============================================================================
# SPRINT HARNESS: Deterministic 72-Hour Engineering Task Runner
# Ingests: SprintPlanArtifact JSON
# Enforces: Task isolation, Phase gates, Kill Switch halting, and Audit logging
# ==============================================================================

set -o pipefail

PLAN_FILE="${1:-sprint_plan.json}"
LOG_DIR="./sprint_run_logs"
TIMESTAMP="$(date +'%Y%m%d_%H%M%S')"
RUN_LOG="${LOG_DIR}/sprint_execution_${TIMESTAMP}.log"

mkdir -p "${LOG_DIR}"
exec > >(tee -a "${RUN_LOG}") 2>&1

# ------------------------------------------------------------------------------
# Logging Utilities
# ------------------------------------------------------------------------------
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color

log_info()    { echo -e "${BLUE}[INFO]${NC} $(date +'%H:%M:%S') $1"; }
log_success() { echo -e "${GREEN}[PASS]${NC} $(date +'%H:%M:%S') $1"; }
log_warn()    { echo -e "${YELLOW}[WARN]${NC} $(date +'%H:%M:%S') $1"; }
log_error()   { echo -e "${RED}[FAIL]${NC} $(date +'%H:%M:%S') $1"; }

# ------------------------------------------------------------------------------
# Preflight Validations
# ------------------------------------------------------------------------------
if ! command -v jq &> /dev/null; then
    log_error "Missing dependency: 'jq' is required to parse the sprint artifact."
    exit 1
fi

if [[ ! -f "${PLAN_FILE}" ]]; then
    log_error "Artifact file not found: ${PLAN_FILE}"
    exit 1
fi

if ! jq empty "${PLAN_FILE}" > /dev/null 2>&1; then
    log_error "Invalid JSON formatting in ${PLAN_FILE}"
    exit 1
fi

# ------------------------------------------------------------------------------
# Extract Global Metadata
# ------------------------------------------------------------------------------
CANDIDATE_ID=$(jq -r '.candidate_id // "UNKNOWN"' "${PLAN_FILE}")
CODENAME=$(jq -r '.sprint_codename // "UNNAMED"' "${PLAN_FILE}")
KILL_SWITCH=$(jq -r '.kill_switch_condition // "None specified"' "${PLAN_FILE}")
STACK=$(jq -r '.architecture_summary.runtime_stack // "Generic"' "${PLAN_FILE}")
PATTERN=$(jq -r '.architecture_summary.integration_pattern // "Generic"' "${PLAN_FILE}")

echo "========================================================================"
echo " SPRINT RUNNER: ${CODENAME} (ID: ${CANDIDATE_ID})"
echo " Stack: ${STACK} | Pattern: ${PATTERN}"
echo " Kill Switch Condition: ${KILL_SWITCH}"
echo " Log Destination: ${RUN_LOG}"
echo "========================================================================"

# ------------------------------------------------------------------------------
# Kill Switch Invocation Handler
# ------------------------------------------------------------------------------
trigger_kill_switch() {
    local failing_phase="$1"
    local failing_task="$2"
    local exit_code="$3"

    echo ""
    log_error "****************************************************************"
    log_error "CRITICAL: KILL SWITCH TRIGGERED IN ${failing_phase}"
    log_error "Failing Task: ${failing_task}"
    log_error "Exit Code: ${exit_code}"
    log_error "Monitored Condition: ${KILL_SWITCH}"
    log_error "Sprint terminated immediately to avoid speculative sunk cost."
    log_error "****************************************************************"
    
    # Optional: Append failure context to registry/telemetry
    cat <<EOF > "${LOG_DIR}/kill_switch_event_${TIMESTAMP}.json"
{
  "timestamp": "${TIMESTAMP}",
  "candidate_id": "${CANDIDATE_ID}",
  "codename": "${CODENAME}",
  "failing_phase": "${failing_phase}",
  "failing_task": "${failing_task}",
  "kill_switch_condition": "${KILL_SWITCH}",
  "exit_code": ${exit_code}
}
EOF
    exit 2
}

# ------------------------------------------------------------------------------
# Sequential Execution Loop
# ------------------------------------------------------------------------------
NUM_PHASES=$(jq '.phases | length' "${PLAN_FILE}")

for (( p=0; p<NUM_PHASES; p++ )); do
    PHASE_WINDOW=$(jq -r ".phases[$p].phase_window" "${PLAN_FILE}")
    PHASE_OBJ=$(jq -r ".phases[$p].objective" "${PLAN_FILE}")
    MILESTONE_GATE=$(jq -r ".phases[$p].milestone_gate" "${PLAN_FILE}")

    echo ""
    log_info "--------------------------------------------------------------------"
    log_info "STARTING PHASE: ${PHASE_WINDOW} - ${PHASE_OBJ}"
    log_info "Milestone Gate Requirement: ${MILESTONE_GATE}"
    log_info "--------------------------------------------------------------------"

    NUM_TASKS=$(jq ".phases[$p].tasks | length" "${PLAN_FILE}")

    for (( t=0; t<NUM_TASKS; t++ )); do
        TASK_ID=$(jq -r ".phases[$p].tasks[$t].task_id" "${PLAN_FILE}")
        TASK_TITLE=$(jq -r ".phases[$p].tasks[$t].title" "${PLAN_FILE}")
        VERIFY_CMD=$(jq -r ".phases[$p].tasks[$t].verification_command" "${PLAN_FILE}")
        CRITERIA=$(jq -r ".phases[$p].tasks[$t].acceptance_criteria" "${PLAN_FILE}")

        echo ""
        log_info "[Task ${TASK_ID}] ${TASK_TITLE}"
        log_info "Criteria: ${CRITERIA}"
        log_info "Running Command: ${VERIFY_CMD}"

        # Execute command inside a subshell
        START_TIME=$(date +%s)
        
        bash -c "${VERIFY_CMD}"
        CMD_STATUS=$?
        
        END_TIME=$(date +%s)
        DURATION=$((END_TIME - START_TIME))

        if [[ ${CMD_STATUS} -ne 0 ]]; then
            log_error "Task ${TASK_ID} FAILED with status ${CMD_STATUS} (took ${DURATION}s)"
            
            # If in Phase 1 (Hour 0-24), trigger the strict kill switch
            if [[ "${PHASE_WINDOW}" == "0h-24h" ]]; then
                trigger_kill_switch "${PHASE_WINDOW}" "${TASK_ID}" "${CMD_STATUS}"
            else
                log_error "Post-Phase 1 failure encountered. Aborting run."
                exit 1
            fi
        else
            log_success "Task ${TASK_ID} PASSED (took ${DURATION}s)"
        fi
    done

    log_success "Phase Gate Cleared: ${PHASE_WINDOW} (${MILESTONE_GATE})"
done

# ------------------------------------------------------------------------------
# Final End-to-End Smoke Test
# ------------------------------------------------------------------------------
SMOKE_TEST_SCRIPT=$(jq -r '.smoke_test_script // empty' "${PLAN_FILE}")

if [[ -n "${SMOKE_TEST_SCRIPT}" && "${SMOKE_TEST_SCRIPT}" != "null" ]]; then
    echo ""
    log_info "===================================================================="
    log_info "EXECUTING END-TO-END SMOKE TEST SUITE"
    log_info "===================================================================="
    
    SMOKE_EXEC_FILE="${LOG_DIR}/run_smoke_test_${TIMESTAMP}.sh"
    echo "#!/usr/bin/env bash" > "${SMOKE_EXEC_FILE}"
    echo "set -eo pipefail" >> "${SMOKE_EXEC_FILE}"
    echo "${SMOKE_TEST_SCRIPT}" >> "${SMOKE_EXEC_FILE}"
    chmod +x "${SMOKE_EXEC_FILE}"

    "${SMOKE_EXEC_FILE}"
    SMOKE_STATUS=$?

    if [[ ${SMOKE_STATUS} -ne 0 ]]; then
        log_error "End-to-End smoke test failed with exit code ${SMOKE_STATUS}."
        exit 3
    else
        log_success "End-to-End smoke test completed successfully."
    fi
fi

echo ""
echo "========================================================================"
log_success "SPRINT COMPLETE: All milestone gates and tests verified."
echo "Execution Log Saved: ${RUN_LOG}"
echo "========================================================================"
exit 0

Step-by-Step Usage & Integration

# 1. Make executable
chmod +x sprint_harness.sh

# 2. Run against standard agent artifact
./sprint_harness.sh ./sprint_plan.json

Exit Codes Reference

0: Sprint plan executed and all phase gates passed.
1: Standard task failure in Phase 2 or Phase 3.
2: Kill Switch Triggered (Phase 1 failure; runtime halted immediately to prune dead branches).
3: Phase tasks succeeded, but final smoke_test_script failed validation.

Provide a Python application using Google Gemini API, saving results to markdown files

The standalone script below uses the google-genai SDK and Pydantic structured schema enforcement to execute the multi-agent innovation pipeline and save the resulting artifacts to formatted, timestamped Markdown files.

Script Architecture

                  [Config & API Client Initialization]
                                   │
                                   ▼
                   [Agent 1: Divergence / Generator]
                       Generates candidate idea
                                   │
                                   ▼
                   [Agent 2: ICE Evaluator & Router]
                       Scores Impact / Confidence / Ease
                                   │
                                   ▼
                               Pass / Fail
                                   ├─────────────────────────────┐
                        [If QUICK_WIN]                 [If Non-QuickWin]
                                   │                            │
                                   ▼                            │
                   [Agent 3: Sprint Architect]                  │
                       Breaks into 72h phases                   │
                                   │                            │
                                   └─────────────┬──────────────┘
                                                 ▼
                                     [Markdown Export Engine]
                                 Writes ./reports/YYYYMMDD_*.md

Python Pipeline Implementation (`innovation_pipeline.py`)

#!/usr/bin/env python3
"""
Autonomous Multi-Agent Innovation Pipeline
- Agent 1: Generates novel industrial/user-group candidates.
- Agent 2: Evaluates with calibrated ICE rubric, routes to 4-quadrant portfolio.
- Agent 3: Decomposes accepted Quick Wins into 72-hour engineering sprints.
- Persistence: Exports comprehensive analysis reports to formatted Markdown files.
"""

import os
import sys
from datetime import datetime
from pathlib import Path
from typing import List, Optional
from enum import Enum
from pydantic import BaseModel, Field
from google import genai
from google.genai import types

# ==============================================================================
# 1. Pydantic Structured Data Schemas
# ==============================================================================

class CandidateArtifact(BaseModel):
    candidate_id: str
    concept_name: str
    target_industry: str
    target_user_group: str
    core_mechanism: str
    incumbent_bottleneck: str
    differentiation_factor: str
    primary_failure_mode: str

class PortfolioQuadrant(str, Enum):
    QUICK_WIN = "QUICK_WIN"
    MOONSHOT = "MOONSHOT"
    FILLER = "FILLER"
    DISTRACTION = "DISTRACTION"

class PivotAction(str, Enum):
    NONE = "NONE"
    DECOUPLE_INTERFACE = "DECOUPLE_INTERFACE"
    FEASIBILITY_SPIKE = "FEASIBILITY_SPIKE"
    AGGREGATE_SUBSYSTEMS = "AGGREGATE_SUBSYSTEMS"
    PRUNE_AND_EXCLUDE = "PRUNE_AND_EXCLUDE"

class EvaluationArtifact(BaseModel):
    candidate_id: str
    impact_score: float = Field(ge=1.0, le=10.0)
    confidence_score: float = Field(ge=1.0, le=10.0)
    ease_score: float = Field(ge=1.0, le=10.0)
    effective_ease: float
    composite_ice: float
    quadrant: PortfolioQuadrant
    recommended_action: PivotAction
    impact_justification: str
    feasibility_justification: str
    integration_justification: str
    stress_test_vulnerabilities: List[str]

class SprintTask(BaseModel):
    task_id: str
    title: str
    description: str
    target_artifacts: List[str]
    verification_command: str
    acceptance_criteria: str

class SprintPhase(BaseModel):
    phase_window: str
    objective: str
    tasks: List[SprintTask]
    milestone_gate: str

class SprintPlanArtifact(BaseModel):
    candidate_id: str
    sprint_codename: str
    kill_switch_condition: str
    runtime_stack: str
    integration_pattern: str
    phases: List[SprintPhase]
    smoke_test_script: str

# ==============================================================================
# 2. Pipeline Execution Engine
# ==============================================================================

class InnovationEngine:
    def __init__(self, model_name: str = "gemini-2.5-pro"):
        api_key = os.environ.get("GEMINI_API_KEY")
        if not api_key:
            raise ValueError("GEMINI_API_KEY environment variable is not set.")
        self.client = genai.Client(api_key=api_key)
        self.model_name = model_name
        self.output_dir = Path("./reports")
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def run_agent1_generation(
        self, industry: str, user_group: str, operator_description: str
    ) -> CandidateArtifact:
        """Agent 1: Divergent Generator"""
        prompt = f"""Role: Upstream Innovation Generator (Agent 1)
Target Industry: {industry}
Target User Group: {user_group}
Applied Divergence Operator: {operator_description}

Generate a novel, technically grounded system or process innovation. Focus on root structural bottlenecks rather than superficial digitizations. Return strictly valid JSON.
"""
        response = self.client.models.generate_content(
            model=self.model_name,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.85,
                response_mime_type="application/json",
                response_schema=CandidateArtifact,
            ),
        )
        return CandidateArtifact.model_validate_json(response.text)

    def run_agent2_evaluation(
        self, candidate: CandidateArtifact
    ) -> EvaluationArtifact:
        """Agent 2: Quantitative Evaluator and 4-Quadrant Classifier"""
        prompt = f"""Role: Quantitative ICE Evaluator & Portfolio Gatekeeper (Agent 2)
Evaluate the candidate below using objective criteria (1.0 to 10.0 scale):
- Impact (1-10): Throughput leap or fundamental bottleneck removal.
- Confidence (1-10): Physical/architectural feasibility and precedence.
- Ease (1-10): Low integration friction, non-invasive deployment surface.

Candidate Payload:
{candidate.model_dump_json(indent=2)}

Compute:
- effective_ease = ease * min(1.0, confidence / 5.0)
- composite_ice = (impact * confidence * ease) / 10.0

Classify into Quadrants:
- QUICK_WIN: Impact >= 5.5, Effective Ease >= 5.5
- MOONSHOT: Impact >= 5.5, Effective Ease < 5.5
- FILLER: Impact < 5.5, Effective Ease >= 5.5
- DISTRACTION: Impact < 5.5, Effective Ease < 5.5

Assign Action:
- NONE (Quick Win)
- DECOUPLE_INTERFACE or FEASIBILITY_SPIKE (Moonshot)
- AGGREGATE_SUBSYSTEMS (Filler)
- PRUNE_AND_EXCLUDE (Distraction)
"""
        response = self.client.models.generate_content(
            model=self.model_name,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.1,  # Deterministic evaluation
                response_mime_type="application/json",
                response_schema=EvaluationArtifact,
            ),
        )
        return EvaluationArtifact.model_validate_json(response.text)

    def run_agent3_sprint_planner(
        self, candidate: CandidateArtifact, evaluation: EvaluationArtifact
    ) -> SprintPlanArtifact:
        """Agent 3: 72-Hour Rapid Sprint Architect"""
        prompt = f"""Role: Sprint Architect & Staff Engineer (Agent 3)
Convert this validated QUICK_WIN into an executable 72-hour engineering sprint plan across 3 discrete phases (0h-24h, 24h-48h, 48h-72h).
Enforce: Zero heavy infrastructure, non-invasive deployment, automated verification commands, and an explicit Kill Switch condition for Phase 1.

Candidate:
{candidate.model_dump_json(indent=2)}

Evaluation Metrics:
{evaluation.model_dump_json(indent=2)}
"""
        response = self.client.models.generate_content(
            model=self.model_name,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.2,
                response_mime_type="application/json",
                response_schema=SprintPlanArtifact,
            ),
        )
        return SprintPlanArtifact.model_validate_json(response.text)

    def export_to_markdown(
        self,
        candidate: CandidateArtifact,
        evaluation: EvaluationArtifact,
        sprint: Optional[SprintPlanArtifact] = None,
    ) -> Path:
        """Serializes pipeline results into a structured Markdown file."""
        now = datetime.now()
        timestamp_str = now.strftime("%Y%m%d_%H%M%S")
        slug = candidate.concept_name.lower().replace(" ", "_")[:30]
        file_path = self.output_dir / f"{timestamp_str}_{slug}.md"

        content = []
        content.append(f"# Innovation Opportunity Report: {candidate.concept_name}\n")
        content.append(f"- **Generated:** {now.strftime('%Y-%m-%d %H:%M:%S')}")
        content.append(f"- **Candidate ID:** `{candidate.candidate_id}`")
        content.append(f"- **Industry:** {candidate.target_industry}")
        content.append(f"- **User Group:** {candidate.target_user_group}\n")

        content.append("## 1. Candidate Specification (Agent 1 Divergence)")
        content.append(f"**Core Mechanism:**\n> {candidate.core_mechanism}\n")
        content.append(f"**Target Bottleneck:**\n{candidate.incumbent_bottleneck}\n")
        content.append(f"**Differentiation Factor:**\n{candidate.differentiation_factor}\n")
        content.append(f"**Primary Failure Mode:**\n{candidate.primary_failure_mode}\n")

        content.append("## 2. Quantitative ICE Evaluation (Agent 2 Gatekeeper)")
        content.append("| Metric | Score | Calibration Justification |")
        content.append("| :--- | :---: | :--- |")
        content.append(f"| **Impact** | {evaluation.impact_score:.1f} / 10 | {evaluation.impact_justification} |")
        content.append(f"| **Confidence** | {evaluation.confidence_score:.1f} / 10 | {evaluation.feasibility_justification} |")
        content.append(f"| **Ease** | {evaluation.ease_score:.1f} / 10 | {evaluation.integration_justification} |")
        content.append(f"| **Effective Ease** | {evaluation.effective_ease:.2f} / 10 | Damped by Confidence factor $\\min(1.0, C / 5.0)$ |")
        content.append(f"| **Composite ICE** | {evaluation.composite_ice:.2f} | $(I \\times C \\times E) / 10.0$ |\n")

        content.append(f"**Assigned Portfolio Quadrant:** `{evaluation.quadrant.value}`  ")
        content.append(f"**Recommended Action:** `{evaluation.recommended_action.value}`\n")

        content.append("### Identified Systemic Vulnerabilities")
        for vuln in evaluation.stress_test_vulnerabilities:
            content.append(f"- {vuln}")
        content.append("")

        if sprint:
            content.append("## 3. 72-Hour Engineering Sprint Plan (Agent 3 Orchestrator)")
            content.append(f"- **Codename:** `{sprint.sprint_codename}`")
            content.append(f"- **Runtime Stack:** {sprint.runtime_stack}")
            content.append(f"- **Integration Pattern:** {sprint.integration_pattern}")
            content.append(f"- **Hour 24 Kill Switch:** `{sprint.kill_switch_condition}`\n")

            for phase in sprint.phases:
                content.append(f"### Phase: {phase.phase_window} — {phase.objective}")
                content.append(f"**Milestone Gate:** {phase.milestone_gate}\n")
                content.append("| Task ID | Title & Description | Verification Command | Acceptance Criteria |")
                content.append("| :--- | :--- | :--- | :--- |")
                for task in phase.tasks:
                    content.append(
                        f"| `{task.task_id}` | **{task.title}**<br>{task.description} | `{task.verification_command}` | {task.acceptance_criteria} |"
                    )
                content.append("")

            content.append("### End-to-End Smoke Test Script")
            content.append("```bash")
            content.append(sprint.smoke_test_script.strip())
            content.append("```\n")

        with open(file_path, "w", encoding="utf-8") as f:
            f.write("\n".join(content))

        return file_path

# ==============================================================================
# 3. Main Execution Entry Point
# ==============================================================================

if __name__ == "__main__":
    if "GEMINI_API_KEY" not in os.environ:
        print("Error: GEMINI_API_KEY environment variable is not set.", file=sys.stderr)
        sys.exit(1)

    engine = InnovationEngine(model_name="gemini-2.5-pro")

    print("[1/3] Running Agent 1 (Divergent Innovation Generator)...")
    candidate = engine.run_agent1_generation(
        industry="Spacecraft Flight Software Verification & Safety",
        user_group="Software Safety Assurance Engineers",
        operator_description="Shadow Workflow Formalization (analyzing manual cross-referencing between hazard reports and standard specifications)",
    )
    print(f"      Generated Candidate: '{candidate.concept_name}'")

    print("[2/3] Running Agent 2 (ICE Evaluator & Portfolio Gatekeeper)...")
    evaluation = engine.run_agent2_evaluation(candidate)
    print(f"      Assigned Quadrant: {evaluation.quadrant.value} (Composite ICE: {evaluation.composite_ice})")

    sprint_plan = None
    if evaluation.quadrant == PortfolioQuadrant.QUICK_WIN:
        print("[3/3] Fast-Track Triggered: Running Agent 3 (Sprint Orchestrator)...")
        sprint_plan = engine.run_agent3_sprint_planner(candidate, evaluation)
        print(f"      Sprint Codename: '{sprint_plan.sprint_codename}'")
    else:
        print(f"[3/3] Candidate classified as {evaluation.quadrant.value}. Action: {evaluation.recommended_action.value}. Skipping 72h sprint.")

    saved_path = engine.export_to_markdown(candidate, evaluation, sprint_plan)
    print(f"\n[Done] Innovation report successfully written to: {saved_path}")

Setup & Execution Instructions

# 1. Install prerequisites
pip install google-genai pydantic

# 2. Export your Gemini API key
export GEMINI_API_KEY="your-api-key-here"

# 3. Execute the pipeline
python innovation_pipeline.py

Generated Markdown Structure

The generated file in ./reports/ contains:

Candidate Specification: Explicit candidate ID, root mechanism, bottleneck details, differentiation logic, and primary failure modes.
Quantitative ICE Matrix: Clean Markdown table detailing the 1–10 scores for Impact, Confidence, and Ease, alongside Effective Ease and the composite score.
Execution Routing: Portfolio quadrant classification, recommended action, and identified failure vectors.
Sprint Roadmap (Conditional): If classified as QUICK_WIN, includes the 72-hour breakdown (Phase 1, 2, and 3), task verification commands, milestone gates, and the end-to-end smoke test script.

Modify the script to persist the candidate entities, ICE scores, and sprint plans into a local SQLite database alongside the Markdown reports.

This updated script adds a normalized SQLite persistence layer with foreign key cascading, stores raw phase lists as structured JSON strings, and writes to both SQLite and Markdown in a single atomic transaction per candidate.

#!/usr/bin/env python3
"""
Autonomous Multi-Agent Innovation Pipeline with SQLite & Markdown Persistence
- Agent 1: Generates novel industrial/user-group candidates.
- Agent 2: Evaluates with calibrated ICE rubric, routes to 4-quadrant portfolio.
- Agent 3: Decomposes accepted Quick Wins into 72-hour engineering sprints.
- Persistence:
    * SQLite database (tables: candidates, evaluations, sprint_plans).
    * Formatted, timestamped Markdown reports in ./reports/.
"""

import json
import os
import sqlite3
import sys
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import List, Optional

from google import genai
from google.genai import types
from pydantic import BaseModel, Field

# ==============================================================================
# 1. Pydantic Structured Data Schemas
# ==============================================================================

class CandidateArtifact(BaseModel):
    candidate_id: str
    concept_name: str
    target_industry: str
    target_user_group: str
    core_mechanism: str
    incumbent_bottleneck: str
    differentiation_factor: str
    primary_failure_mode: str

class PortfolioQuadrant(str, Enum):
    QUICK_WIN = "QUICK_WIN"
    MOONSHOT = "MOONSHOT"
    FILLER = "FILLER"
    DISTRACTION = "DISTRACTION"

class PivotAction(str, Enum):
    NONE = "NONE"
    DECOUPLE_INTERFACE = "DECOUPLE_INTERFACE"
    FEASIBILITY_SPIKE = "FEASIBILITY_SPIKE"
    AGGREGATE_SUBSYSTEMS = "AGGREGATE_SUBSYSTEMS"
    PRUNE_AND_EXCLUDE = "PRUNE_AND_EXCLUDE"

class EvaluationArtifact(BaseModel):
    candidate_id: str
    impact_score: float = Field(ge=1.0, le=10.0)
    confidence_score: float = Field(ge=1.0, le=10.0)
    ease_score: float = Field(ge=1.0, le=10.0)
    effective_ease: float
    composite_ice: float
    quadrant: PortfolioQuadrant
    recommended_action: PivotAction
    impact_justification: str
    feasibility_justification: str
    integration_justification: str
    stress_test_vulnerabilities: List[str]

class SprintTask(BaseModel):
    task_id: str
    title: str
    description: str
    target_artifacts: List[str]
    verification_command: str
    acceptance_criteria: str

class SprintPhase(BaseModel):
    phase_window: str
    objective: str
    tasks: List[SprintTask]
    milestone_gate: str

class SprintPlanArtifact(BaseModel):
    candidate_id: str
    sprint_codename: str
    kill_switch_condition: str
    runtime_stack: str
    integration_pattern: str
    phases: List[SprintPhase]
    smoke_test_script: str

# ==============================================================================
# 2. SQLite Database Manager
# ==============================================================================

class InnovationDatabase:
    """Handles DDL schema initialization and transactional persistence."""

    def __init__(self, db_path: str = "innovation_portfolio.db"):
        self.db_path = db_path
        self._init_schema()

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.execute("PRAGMA foreign_keys = ON;")
        conn.row_factory = sqlite3.Row
        return conn

    def _init_schema(self) -> None:
        """Initializes tables for candidates, evaluations, and sprint plans."""
        schema = """
        CREATE TABLE IF NOT EXISTS candidates (
            candidate_id TEXT PRIMARY KEY,
            concept_name TEXT NOT NULL,
            target_industry TEXT NOT NULL,
            target_user_group TEXT NOT NULL,
            core_mechanism TEXT NOT NULL,
            incumbent_bottleneck TEXT NOT NULL,
            differentiation_factor TEXT NOT NULL,
            primary_failure_mode TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS evaluations (
            evaluation_id INTEGER PRIMARY KEY AUTOINCREMENT,
            candidate_id TEXT NOT NULL,
            impact_score REAL NOT NULL,
            confidence_score REAL NOT NULL,
            ease_score REAL NOT NULL,
            effective_ease REAL NOT NULL,
            composite_ice REAL NOT NULL,
            quadrant TEXT NOT NULL,
            recommended_action TEXT NOT NULL,
            impact_justification TEXT,
            feasibility_justification TEXT,
            integration_justification TEXT,
            stress_test_vulnerabilities TEXT,
            evaluated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (candidate_id) REFERENCES candidates(candidate_id) ON DELETE CASCADE
        );

        CREATE TABLE IF NOT EXISTS sprint_plans (
            sprint_id INTEGER PRIMARY KEY AUTOINCREMENT,
            candidate_id TEXT NOT NULL,
            sprint_codename TEXT NOT NULL,
            kill_switch_condition TEXT NOT NULL,
            runtime_stack TEXT NOT NULL,
            integration_pattern TEXT NOT NULL,
            smoke_test_script TEXT NOT NULL,
            phases_json TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (candidate_id) REFERENCES candidates(candidate_id) ON DELETE CASCADE
        );
        """
        with self._get_connection() as conn:
            conn.executescript(schema)

    def persist_run(
        self,
        candidate: CandidateArtifact,
        evaluation: EvaluationArtifact,
        sprint: Optional[SprintPlanArtifact] = None,
    ) -> None:
        """Inserts candidate, evaluation, and optional sprint records atomically."""
        with self._get_connection() as conn:
            cur = conn.cursor()

            # 1. Upsert / Insert Candidate
            cur.execute(
                """
                INSERT INTO candidates (
                    candidate_id, concept_name, target_industry, target_user_group,
                    core_mechanism, incumbent_bottleneck, differentiation_factor, primary_failure_mode
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(candidate_id) DO UPDATE SET
                    concept_name=excluded.concept_name,
                    core_mechanism=excluded.core_mechanism,
                    incumbent_bottleneck=excluded.incumbent_bottleneck,
                    differentiation_factor=excluded.differentiation_factor,
                    primary_failure_mode=excluded.primary_failure_mode;
                """,
                (
                    candidate.candidate_id,
                    candidate.concept_name,
                    candidate.target_industry,
                    candidate.target_user_group,
                    candidate.core_mechanism,
                    candidate.incumbent_bottleneck,
                    candidate.differentiation_factor,
                    candidate.primary_failure_mode,
                ),
            )

            # 2. Insert Evaluation
            cur.execute(
                """
                INSERT INTO evaluations (
                    candidate_id, impact_score, confidence_score, ease_score,
                    effective_ease, composite_ice, quadrant, recommended_action,
                    impact_justification, feasibility_justification, integration_justification,
                    stress_test_vulnerabilities
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    evaluation.candidate_id,
                    evaluation.impact_score,
                    evaluation.confidence_score,
                    evaluation.ease_score,
                    evaluation.effective_ease,
                    evaluation.composite_ice,
                    evaluation.quadrant.value,
                    evaluation.recommended_action.value,
                    evaluation.impact_justification,
                    evaluation.feasibility_justification,
                    evaluation.integration_justification,
                    json.dumps(evaluation.stress_test_vulnerabilities),
                ),
            )

            # 3. Insert Sprint Plan (if admitted as Quick Win)
            if sprint:
                phases_serialized = json.dumps([p.model_dump() for p in sprint.phases])
                cur.execute(
                    """
                    INSERT INTO sprint_plans (
                        candidate_id, sprint_codename, kill_switch_condition,
                        runtime_stack, integration_pattern, smoke_test_script, phases_json
                    ) VALUES (?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        sprint.candidate_id,
                        sprint.sprint_codename,
                        sprint.kill_switch_condition,
                        sprint.runtime_stack,
                        sprint.integration_pattern,
                        sprint.smoke_test_script,
                        phases_serialized,
                    ),
                )

# ==============================================================================
# 3. Pipeline Execution Engine
# ==============================================================================

class InnovationEngine:
    def __init__(
        self,
        model_name: str = "gemini-2.5-pro",
        db_path: str = "innovation_portfolio.db",
    ):
        api_key = os.environ.get("GEMINI_API_KEY")
        if not api_key:
            raise ValueError("GEMINI_API_KEY environment variable is not set.")
        self.client = genai.Client(api_key=api_key)
        self.model_name = model_name
        self.output_dir = Path("./reports")
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.db = InnovationDatabase(db_path=db_path)

    def run_agent1_generation(
        self, industry: str, user_group: str, operator_description: str
    ) -> CandidateArtifact:
        """Agent 1: Divergent Generator"""
        prompt = f"""Role: Upstream Innovation Generator (Agent 1)
Target Industry: {industry}
Target User Group: {user_group}
Applied Divergence Operator: {operator_description}

Generate a novel, technically grounded system or process innovation. Focus on root structural bottlenecks rather than superficial digitizations. Return strictly valid JSON.
"""
        response = self.client.models.generate_content(
            model=self.model_name,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.85,
                response_mime_type="application/json",
                response_schema=CandidateArtifact,
            ),
        )
        return CandidateArtifact.model_validate_json(response.text)

    def run_agent2_evaluation(
        self, candidate: CandidateArtifact
    ) -> EvaluationArtifact:
        """Agent 2: Quantitative Evaluator and 4-Quadrant Classifier"""
        prompt = f"""Role: Quantitative ICE Evaluator & Portfolio Gatekeeper (Agent 2)
Evaluate the candidate below using objective criteria (1.0 to 10.0 scale):
- Impact (1-10): Throughput leap or fundamental bottleneck removal.
- Confidence (1-10): Physical/architectural feasibility and precedence.
- Ease (1-10): Low integration friction, non-invasive deployment surface.

Candidate Payload:
{candidate.model_dump_json(indent=2)}

Compute:
- effective_ease = ease * min(1.0, confidence / 5.0)
- composite_ice = (impact * confidence * ease) / 10.0

Classify into Quadrants:
- QUICK_WIN: Impact >= 5.5, Effective Ease >= 5.5
- MOONSHOT: Impact >= 5.5, Effective Ease < 5.5
- FILLER: Impact < 5.5, Effective Ease >= 5.5
- DISTRACTION: Impact < 5.5, Effective Ease < 5.5

Assign Action:
- NONE (Quick Win)
- DECOUPLE_INTERFACE or FEASIBILITY_SPIKE (Moonshot)
- AGGREGATE_SUBSYSTEMS (Filler)
- PRUNE_AND_EXCLUDE (Distraction)
"""
        response = self.client.models.generate_content(
            model=self.model_name,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.1,
                response_mime_type="application/json",
                response_schema=EvaluationArtifact,
            ),
        )
        return EvaluationArtifact.model_validate_json(response.text)

    def run_agent3_sprint_planner(
        self, candidate: CandidateArtifact, evaluation: EvaluationArtifact
    ) -> SprintPlanArtifact:
        """Agent 3: 72-Hour Rapid Sprint Architect"""
        prompt = f"""Role: Sprint Architect & Staff Engineer (Agent 3)
Convert this validated QUICK_WIN into an executable 72-hour engineering sprint plan across 3 discrete phases (0h-24h, 24h-48h, 48h-72h).
Enforce: Zero heavy infrastructure, non-invasive deployment, automated verification commands, and an explicit Kill Switch condition for Phase 1.

Candidate:
{candidate.model_dump_json(indent=2)}

Evaluation Metrics:
{evaluation.model_dump_json(indent=2)}
"""
        response = self.client.models.generate_content(
            model=self.model_name,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.2,
                response_mime_type="application/json",
                response_schema=SprintPlanArtifact,
            ),
        )
        return SprintPlanArtifact.model_validate_json(response.text)

    def export_to_markdown(
        self,
        candidate: CandidateArtifact,
        evaluation: EvaluationArtifact,
        sprint: Optional[SprintPlanArtifact] = None,
    ) -> Path:
        """Serializes pipeline results into a structured Markdown file."""
        now = datetime.now()
        timestamp_str = now.strftime("%Y%m%d_%H%M%S")
        slug = candidate.concept_name.lower().replace(" ", "_")[:30]
        file_path = self.output_dir / f"{timestamp_str}_{slug}.md"

        content = []
        content.append(f"# Innovation Opportunity Report: {candidate.concept_name}\n")
        content.append(f"- **Generated:** {now.strftime('%Y-%m-%d %H:%M:%S')}")
        content.append(f"- **Candidate ID:** `{candidate.candidate_id}`")
        content.append(f"- **Industry:** {candidate.target_industry}")
        content.append(f"- **User Group:** {candidate.target_user_group}\n")

        content.append("## 1. Candidate Specification (Agent 1 Divergence)")
        content.append(f"**Core Mechanism:**\n> {candidate.core_mechanism}\n")
        content.append(f"**Target Bottleneck:**\n{candidate.incumbent_bottleneck}\n")
        content.append(f"**Differentiation Factor:**\n{candidate.differentiation_factor}\n")
        content.append(f"**Primary Failure Mode:**\n{candidate.primary_failure_mode}\n")

        content.append("## 2. Quantitative ICE Evaluation (Agent 2 Gatekeeper)")
        content.append("| Metric | Score | Calibration Justification |")
        content.append("| :--- | :---: | :--- |")
        content.append(f"| **Impact** | {evaluation.impact_score:.1f} / 10 | {evaluation.impact_justification} |")
        content.append(f"| **Confidence** | {evaluation.confidence_score:.1f} / 10 | {evaluation.feasibility_justification} |")
        content.append(f"| **Ease** | {evaluation.ease_score:.1f} / 10 | {evaluation.integration_justification} |")
        content.append(f"| **Effective Ease** | {evaluation.effective_ease:.2f} / 10 | Damped by Confidence factor $\\min(1.0, C / 5.0)$ |")
        content.append(f"| **Composite ICE** | {evaluation.composite_ice:.2f} | $(I \\times C \\times E) / 10.0$ |\n")

        content.append(f"**Assigned Portfolio Quadrant:** `{evaluation.quadrant.value}`  ")
        content.append(f"**Recommended Action:** `{evaluation.recommended_action.value}`\n")

        content.append("### Identified Systemic Vulnerabilities")
        for vuln in evaluation.stress_test_vulnerabilities:
            content.append(f"- {vuln}")
        content.append("")

        if sprint:
            content.append("## 3. 72-Hour Engineering Sprint Plan (Agent 3 Orchestrator)")
            content.append(f"- **Codename:** `{sprint.sprint_codename}`")
            content.append(f"- **Runtime Stack:** {sprint.runtime_stack}")
            content.append(f"- **Integration Pattern:** {sprint.integration_pattern}")
            content.append(f"- **Hour 24 Kill Switch:** `{sprint.kill_switch_condition}`\n")

            for phase in sprint.phases:
                content.append(f"### Phase: {phase.phase_window} — {phase.objective}")
                content.append(f"**Milestone Gate:** {phase.milestone_gate}\n")
                content.append("| Task ID | Title & Description | Verification Command | Acceptance Criteria |")
                content.append("| :--- | :--- | :--- | :--- |")
                for task in phase.tasks:
                    content.append(
                        f"| `{task.task_id}` | **{task.title}**<br>{task.description} | `{task.verification_command}` | {task.acceptance_criteria} |"
                    )
                content.append("")

            content.append("### End-to-End Smoke Test Script")
            content.append("```bash")
            content.append(sprint.smoke_test_script.strip())
            content.append("```\n")

        with open(file_path, "w", encoding="utf-8") as f:
            f.write("\n".join(content))

        return file_path

    def run_pipeline(
        self, industry: str, user_group: str, operator_description: str
    ) -> None:
        """Executes full agent flow, commits to SQLite, and writes Markdown."""
        print("[1/3] Running Agent 1 (Divergent Innovation Generator)...")
        candidate = self.run_agent1_generation(industry, user_group, operator_description)
        print(f"      Candidate: '{candidate.concept_name}' (ID: {candidate.candidate_id})")

        print("[2/3] Running Agent 2 (ICE Evaluator & Portfolio Gatekeeper)...")
        evaluation = self.run_agent2_evaluation(candidate)
        print(f"      Quadrant: {evaluation.quadrant.value} (Composite ICE: {evaluation.composite_ice})")

        sprint_plan = None
        if evaluation.quadrant == PortfolioQuadrant.QUICK_WIN:
            print("[3/3] Fast-Track Triggered: Running Agent 3 (Sprint Orchestrator)...")
            sprint_plan = self.run_agent3_sprint_planner(candidate, evaluation)
            print(f"      Sprint Codename: '{sprint_plan.sprint_codename}'")
        else:
            print(f"[3/3] Candidate classified as {evaluation.quadrant.value}. Action: {evaluation.recommended_action.value}.")

        # Dual Persistence
        self.db.persist_run(candidate, evaluation, sprint_plan)
        print(f"      [SQLite] Successfully recorded in '{self.db.db_path}'")

        report_path = self.export_to_markdown(candidate, evaluation, sprint_plan)
        print(f"      [Markdown] Report generated: {report_path}")

# ==============================================================================
# 4. Entry Point
# ==============================================================================

if __name__ == "__main__":
    if "GEMINI_API_KEY" not in os.environ:
        print("Error: GEMINI_API_KEY environment variable is not set.", file=sys.stderr)
        sys.exit(1)

    engine = InnovationEngine(
        model_name="gemini-2.5-pro",
        db_path="innovation_portfolio.db",
    )

    engine.run_pipeline(
        industry="Spacecraft Flight Software Verification & Safety",
        user_group="Software Safety Assurance Engineers",
        operator_description="Shadow Workflow Formalization (analyzing manual cross-referencing between hazard reports and standard specifications)",
    )

Verifying SQLite Contents

Inspect the database directly with Python or the SQLite CLI:

import sqlite3

conn = sqlite3.connect("innovation_portfolio.db")
conn.row_factory = sqlite3.Row
cur = conn.cursor()

query = """
SELECT 
    c.concept_name,
    e.quadrant,
    e.composite_ice,
    s.sprint_codename
FROM candidates c
JOIN evaluations e ON c.candidate_id = e.candidate_id
LEFT JOIN sprint_plans s ON c.candidate_id = s.candidate_id
ORDER BY c.created_at DESC;
"""

for row in cur.execute(query):
    print(dict(row))

From <https://gemini.google.com/app/751ed735648c2c7d> Google Gemini (3.8 Flash)