#4934 Ask Chat: Improved Prompt, Recommendation and Implementation Agent Workflow
#4934 Ask Chat: Improved Prompt, Recommendation and Implementation Agent Workflow
Here are three ways to improve the problem statement, depending on whether you need an executive-level summary, a formal software requirements specification, or an architectural workflow description.
Option 1: Formal & Concise (Recommended for Specs & Proposals)
Objective: Design and implement a two-stage multi-agent orchestration workflow that transforms raw user prompts into executable technical plans and automated implementations.
Key Capabilities:
Problem Refinement: Ingest, parse, and systematically disambiguate the initial user problem statement.
Stage 1 (Planning Orchestrator): Dynamically inspect a file-based skill library, select relevant domain skills, and orchestrate specialized agents to produce a comprehensive execution strategy.
Stage 2 (Implementation Orchestrator): Ingest the generated plan, dynamically resolve the operational skills required for execution, and coordinate execution agents to implement the solution.
Option 2: Executive Summary (Problem / Solution Format)
Current Challenge: Turning unstructured, ambiguous problem statements into production-ready implementations typically requires manual problem framing, plan definition, and coordination across disparate tools and skills.
Proposed Solution: Build an automated end-to-end multi-agent orchestration pipeline featuring:
Intake & Refinement: An intake module that sharpens and clarifies the user-defined problem.
Strategic Orchestration: A planner orchestrator that queries a local file/folder skill repository to assign task-specific skills and draft a detailed execution plan.
Tactical Orchestration: A downstream execution orchestrator that receives the plan, provisions specialized agents with the required tools/skills, and executes the implementation deliverables.
Option 3: Architectural Workflow Breakdown (For Engineering/PRD)
[User Input] │ ▼ [Refinement Engine] ── Clarifies & refines scope │ ▼ [Planning Orchestrator] ◄──► [File/Directory Skill Library] │ (Matches skills to problem requirements) ▼ [Structured Execution Plan] │ ▼ [Execution Orchestrator] ◄──► [Operational Skill Library] │ (Binds execution skills & agents) ▼ [Final Implementation / Artifacts]
Key Improvements Made Over the Original Draft
Clarified the Separation of Concerns: Divided the pipeline into distinct Strategic Planning (Stage 1) and Operational Execution (Stage 2) phases.
Standardized Agent Terminology: Replaced passive phrasing with explicit operational terms like skill binding/resolution, ingest & disambiguate, and multi-agent coordination.
Precision on the Skill Library: Clarified that the skill library acts as a discoverable repository for dynamic agent assignment.
Design a file and folder directory structure and metadata schema for the agent skill library.
Directory Architecture
A file-based skill library relies on a modular, self-contained structure where each skill encapsulates its instructions, schemas, executable logic, and auxiliary resources. Organizing skills by operational domain (e.g., planning/, implementation/, verification/) allows dynamic discovery and progressive disclosure without loading extraneous context into agent prompts.
skills_library/ ├── registry.json # Auto-generated or managed global index ├── _common/ # Shared base models, utilities, and types │ ├── base_models.py # Pydantic base classes for inputs/outputs │ └── error_handlers.py # Standardized exception definitions │ ├── planning/ # Stage 1: Problem Refinement & Planning Skills │ ├── problem_refiner/ │ │ ├── SKILL.md # YAML frontmatter + prompt/instruction body │ │ ├── schemas.py # Pydantic Input/Output contracts │ │ ├── handler.py # Optional deterministic pre/post-processing │ │ └── examples/ │ │ ├── input_sample.json │ │ └── output_sample.json │ └── task_decomposer/ │ ├── SKILL.md │ ├── schemas.py │ └── handler.py │ ├── implementation/ # Stage 2: Code, Config & Artifact Execution │ ├── python_scaffolder/ │ │ ├── SKILL.md │ │ ├── schemas.py │ │ ├── handler.py │ │ └── templates/ # Domain-specific scaffolding templates │ │ └── module_template.py.jinja │ └── file_writer/ │ ├── SKILL.md │ ├── schemas.py │ └── handler.py │ └── verification/ # Evaluator/Critic & Validation Skills ├── schema_validator/ │ ├── SKILL.md │ ├── schemas.py │ └── handler.py └── plan_compliance_auditor/ ├── SKILL.md ├── schemas.py └── handler.py
Standard SKILL.md Layout
Each SKILL.md file serves as the single source of truth for both the orchestrator (via YAML frontmatter) and the instantiated subagent (via markdown instructions).
---
id: "planning.problem_refiner"
name: "Problem Refiner"
version: "1.2.0"
stage: "planning"
description: "Ingests ambiguous problem statements, identifies hidden assumptions, and produces a structured, actionable problem specification."
tags:
- "refinement"
- "requirements"
- "intake"
compatibility:
min_agent_runtime: "0.4.0"
deterministic_validator: true
contracts:
input_schema: "schemas.RefinerInput"
output_schema: "schemas.RefinerOutput"
execution:
mode: "llm_with_postprocess"
entrypoint: "handler.py:execute"
---
# Problem Refiner Skill
## Role & Mission
You are an expert systems requirements engineer. Your task is to disambiguate, refine, and scope the raw problem statement submitted by the user.
## Analytical Procedure
1. Extract the core business or technical objective.
2. Identify missing constraints, implicit dependencies, and ambiguous terminology.
3. Formulate clarifying assumptions for any undefined parameters.
4. Output the structured refinement conforming strictly to `RefinerOutput`.
## Boundaries & Constraints
- Do not propose implementation code or technical solutions.
- Restrict scope strictly to framing and specifying the problem boundaries.
Pydantic Metadata & Contract Schemas (schemas.py)
A formal metadata schema allows the orchestrator to dynamically index, filter, and hydrate agents without reading entire markdown files into memory.
from enum import Enum
from typing import List, Optional, Dict, Any
from pydantic import BaseModel, Field
class OrchestrationStage(str, Enum):
PLANNING = "planning"
IMPLEMENTATION = "implementation"
VERIFICATION = "verification"
GENERAL = "general"
class ExecutionMode(str, Enum):
LLM_ONLY = "llm_only"
DETERMINISTIC_ONLY = "deterministic_only"
LLM_WITH_POSTPROCESS = "llm_with_postprocess"
class ContractDefinition(BaseModel):
input_schema: str = Field(..., description="Dotpath to Pydantic class, e.g., schemas.RefinerInput")
output_schema: str = Field(..., description="Dotpath to Pydantic class, e.g., schemas.RefinerOutput")
class CompatibilityRules(BaseModel):
min_agent_runtime: str = "0.1.0"
deterministic_validator: bool = True
required_system_tools: List[str] = Field(default_factory=list)
class SkillMetadata(BaseModel):
id: str = Field(..., regex=r"^[a-z_]+(\.[a-z_]+)+$", description="Namespaced unique identifier (e.g., planning.problem_refiner)")
name: str = Field(..., min_length=3, max_length=60)
version: str = Field(..., regex=r"^\d+\.\d+\.\d+$", description="Semantic version string")
stage: OrchestrationStage
description: str = Field(..., min_length=20, max_length=300)
tags: List[str] = Field(default_factory=list)
contracts: ContractDefinition
compatibility: CompatibilityRules = Field(default_factory=CompatibilityRules)
execution: Dict[str, Any] = Field(default_factory=lambda: {"mode": ExecutionMode.LLM_ONLY})
# --- Operational Payload Contracts (Example for problem_refiner) ---
class RefinerInput(BaseModel):
raw_statement: str = Field(..., description="The user's original unparsed problem prompt.")
context_parameters: Optional[Dict[str, Any]] = Field(default_factory=dict)
class RefinedProblemElement(BaseModel):
core_objective: str
in_scope: List[str]
out_of_scope: List[str]
assumptions: List[str]
identified_constraints: List[str]
class RefinerOutput(BaseModel):
success: bool
refined_statement: RefinedProblemElement
recommended_execution_skills: List[str] = Field(
default_factory=list,
description="List of skill IDs suggested for subsequent stages."
)
Dynamic Skill Discovery & Registration Loader
The loader scans skill folders, parses the YAML frontmatter, registers available capabilities, and surfaces metadata to the orchestrator for lightweight planning.
from pathlib import Path
import yaml
from pydantic import ValidationError
class SkillLibraryRegistry:
def __init__(self, root_dir: str):
self.root_dir = Path(root_dir)
self.skills: dict[str, SkillMetadata] = {}
self.instruction_cache: dict[str, str] = {}
def scan_and_index(self) -> None:
"""Walks the folder hierarchy and indexes all SKILL.md definitions."""
self.skills.clear()
self.instruction_cache.clear()
for skill_file in self.root_dir.glob("**/SKILL.md"):
try:
raw_text = skill_file.read_text(encoding="utf-8")
parts = raw_text.split("---", 2)
if len(parts) < 3:
continue # Invalid frontmatter structure
frontmatter_yaml = parts[1]
instructions = parts[2].strip()
meta_dict = yaml.safe_load(frontmatter_yaml)
skill_meta = SkillMetadata(**meta_dict)
self.skills[skill_meta.id] = skill_meta
self.instruction_cache[skill_meta.id] = instructions
except (yaml.YAMLError, ValidationError, Exception) as err:
print(f"[WARN] Failed to parse {skill_file}: {err}")
def query_by_stage(self, stage: OrchestrationStage) -> list[SkillMetadata]:
return [skill for skill in self.skills.values() if skill.stage == stage]
def get_skill_instructions(self, skill_id: str) -> str:
return self.instruction_cache.get(skill_id, "")
Implement the Python code for the Stage 1 Planning Orchestrator that scans this skill library and dynamically binds agents to problem specifications.
Here is the complete implementation for the Stage 1 Planning Orchestrator. It reads the folder-based skill library, exposes metadata via progressive disclosure to select the right skills, hydrates an execution agent with the selected skill's prompt and schema contracts, and produces a structured, actionable plan for Stage 2.
System Architecture
[Raw Problem Input] │ ▼ [SkillLibraryRegistry] ──── Scans & indexes frontmatter (id, description, tags, contracts) │ ▼ [Stage 1 Planning Orchestrator] ├── Step 1: Disambiguate & Refine Problem Statement ├── Step 2: Match & Select Candidate Skills from Registry (Progressive Disclosure) └── Step 3: Hydrate & Dispatch Planning Agent with Selected Skills │ ▼ [Structured Execution Plan (Pydantic)] ──► Ready for Stage 2 Implementation
Core Data Models (schemas.py)
"""schemas.py
Pydantic contracts for skill discovery, orchestrator execution, and plan outputs.
"""
from __future__ import annotations
from enum import Enum
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class OrchestrationStage(str, Enum):
PLANNING = "planning"
IMPLEMENTATION = "implementation"
VERIFICATION = "verification"
GENERAL = "general"
class SkillMetadata(BaseModel):
id: str = Field(..., description="Namespaced ID, e.g. planning.problem_refiner")
name: str
version: str
stage: OrchestrationStage
description: str
tags: List[str] = Field(default_factory=list)
input_schema: str = Field(..., description="Class name for input contract")
output_schema: str = Field(..., description="Class name for output contract")
class RefinedProblem(BaseModel):
core_objective: str = Field(..., description="Primary clear goal")
scope_in: List[str] = Field(default_factory=list)
scope_out: List[str] = Field(default_factory=list)
identified_constraints: List[str] = Field(default_factory=list)
key_assumptions: List[str] = Field(default_factory=list)
class PlannedTask(BaseModel):
task_id: str = Field(..., description="Unique step ID, e.g. TASK-01")
title: str
description: str
target_stage: OrchestrationStage = OrchestrationStage.IMPLEMENTATION
recommended_skills: List[str] = Field(
default_factory=list,
description="Skill IDs required for Stage 2 execution (e.g., implementation.python_scaffolder)"
)
dependencies: List[str] = Field(
default_factory=list,
description="Task IDs that must complete prior to this task"
)
expected_artifact: str = Field(..., description="File path, code bundle, or structured output description")
class Stage1PlanningOutput(BaseModel):
refined_problem: RefinedProblem
execution_plan: List[PlannedTask]
audit_notes: str = Field(..., description="Rationale for selected architecture and skill bindings")
Registry Loader & Orchestrator Implementation (stage1_orchestrator.py)
This module uses standard structured outputs (compatible with the google-genai SDK or Ollama structured outputs via Pydantic).
"""stage1_orchestrator.py
Scans the file-based skill library, performs problem refinement, binds skills,
and generates an execution plan ready for Stage 2.
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Dict, List, Optional
import yaml
from google import genai
from google.genai import types
from schemas import (
OrchestrationStage,
PlannedTask,
RefinedProblem,
SkillMetadata,
Stage1PlanningOutput,
)
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("PlanningOrchestrator")
class SkillLibraryRegistry:
"""Discovers, parses, and indexes SKILL.md bundles across directory trees."""
def __init__(self, root_dir: str | Path):
self.root_dir = Path(root_dir)
self.skills: Dict[str, SkillMetadata] = {}
self.instructions: Dict[str, str] = {}
def scan(self) -> None:
"""Indexes metadata via YAML frontmatter while keeping prompt bodies cached separately."""
self.skills.clear()
self.instructions.clear()
if not self.root_dir.exists():
raise FileNotFoundError(f"Skill library directory not found: {self.root_dir}")
for skill_path in self.root_dir.glob("**/SKILL.md"):
try:
content = skill_path.read_text(encoding="utf-8")
parts = content.split("---", 2)
if len(parts) < 3:
logger.warning("Skipping malformed SKILL.md (missing frontmatter fences): %s", skill_path)
continue
frontmatter = yaml.safe_load(parts[1])
contracts = frontmatter.get("contracts", {})
meta = SkillMetadata(
id=frontmatter["id"],
name=frontmatter["name"],
version=str(frontmatter.get("version", "1.0.0")),
stage=OrchestrationStage(frontmatter.get("stage", "general")),
description=frontmatter["description"],
tags=frontmatter.get("tags", []),
input_schema=contracts.get("input_schema", "BaseModel"),
output_schema=contracts.get("output_schema", "BaseModel"),
)
self.skills[meta.id] = meta
self.instructions[meta.id] = parts[2].strip()
logger.info("Registered skill [%s] from %s", meta.id, skill_path.relative_to(self.root_dir))
except Exception as exc:
logger.error("Failed to load skill at %s: %s", skill_path, exc)
def list_manifests(self, stage: Optional[OrchestrationStage] = None) -> List[Dict[str, Any]]:
"""Returns lightweight capability summaries for context-efficient LLM discovery."""
manifests = []
for s in self.skills.values():
if stage is None or s.stage == stage:
manifests.append({
"id": s.id,
"name": s.name,
"stage": s.stage.value,
"description": s.description,
"tags": s.tags,
})
return manifests
def get_instructions(self, skill_id: str) -> str:
return self.instructions.get(skill_id, "")
class Stage1PlanningOrchestrator:
"""Coordinates problem refinement, skill resolution, and structured plan decomposition."""
def __init__(self, registry: SkillLibraryRegistry, model_id: str = "gemini-2.5-pro"):
self.registry = registry
self.model_id = model_id
self.client = genai.Client()
def run(self, raw_user_prompt: str) -> Stage1PlanningOutput:
"""Executes the complete Stage 1 refinement and planning pipeline."""
logger.info("Beginning Stage 1 Orchestration...")
# 1. Summarize available skills across all stages for the planner
available_skills_catalog = self.registry.list_manifests()
catalog_json = json.dumps(available_skills_catalog, indent=2)
# 2. Check if specific planning skills exist in library (e.g. problem_refiner)
refiner_instructions = self.registry.get_instructions("planning.problem_refiner")
refiner_context = (
f"\n\nSpecialized Planning Instructions Found in Library:\n{refiner_instructions}"
if refiner_instructions
else ""
)
system_instruction = f"""You are the Master Planning & Systems Orchestrator (Stage 1).
Your mission is to take an ambiguous user problem statement, refine it into crisp technical requirements,
and formulate a multi-step execution plan where each task is mapped directly to compatible skills
from the provided Skill Catalog.
### Available Skill Catalog (Metadata Only):
{catalog_json}
{refiner_context}
### Operational Rules:
1. Disambiguate all terminology and separate scope clearly into scope_in and scope_out.
2. Formulate explicit, unstated technical assumptions and constraints.
3. Decompose the implementation into sequential PlannedTasks.
4. For each task's `recommended_skills`, select EXACT `id` matches from the Available Skill Catalog.
5. Ensure task dependencies form an acyclic workflow (DAG).
"""
logger.info("Dispatching synthesis request to LLM with structured schema...")
response = self.client.models.generate_content(
model=self.model_id,
contents=[types.Content(role="user", parts=[types.Part.from_text(raw_user_prompt)])],
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=Stage1PlanningOutput,
temperature=0.1,
),
)
# 3. Parse and validate output
structured_plan: Stage1PlanningOutput = Stage1PlanningOutput.model_validate_json(response.text)
logger.info(
"Stage 1 complete: Formulated %d tasks across %d identified constraints.",
len(structured_plan.execution_plan),
len(structured_plan.refined_problem.identified_constraints),
)
return structured_plan
Verification & Test Harness (run_stage1_demo.py)
This runnable script creates a mock directory tree, tests registry parsing, runs the planner, and formats the output.
"""run_stage1_demo.py
Demonstrates end-to-end execution of the Stage 1 Planning Orchestrator.
"""
import json
import shutil
from pathlib import Path
from stage1_orchestrator import SkillLibraryRegistry, Stage1PlanningOrchestrator
def setup_mock_skill_library(base_path: Path):
"""Generates sample skills to verify dynamic discovery."""
if base_path.exists():
shutil.rmtree(base_path)
# 1. Planning Skill: Problem Refiner
refiner_dir = base_path / "planning" / "problem_refiner"
refiner_dir.mkdir(parents=True, exist_ok=True)
(refiner_dir / "SKILL.md").write_text(
"""---
id: "planning.problem_refiner"
name: "Problem Refiner"
version: "1.1.0"
stage: "planning"
description: "Disambiguates messy problem statements and extracts formal constraints."
tags: ["requirements", "scoping"]
contracts:
input_schema: "RawInput"
output_schema: "RefinedProblem"
---
Refine statements by isolating actors, functional targets, and hard failure states.
""",
encoding="utf-8",
)
# 2. Implementation Skill: Python Scaffolder
py_dir = base_path / "implementation" / "python_scaffolder"
py_dir.mkdir(parents=True, exist_ok=True)
(py_dir / "SKILL.md").write_text(
"""---
id: "implementation.python_scaffolder"
name: "Python Module Scaffolder"
version: "2.0.0"
stage: "implementation"
description: "Generates boilerplate packages, pyproject.toml, and modular folder structures."
tags: ["python", "codegen", "packaging"]
contracts:
input_schema: "ScaffoldSpec"
output_schema: "FileSystemResult"
---
Construct production-ready Python structures adhering strictly to PEP 621.
""",
encoding="utf-8",
)
# 3. Implementation Skill: SQLite Storage Builder
db_dir = base_path / "implementation" / "sqlite_builder"
db_dir.mkdir(parents=True, exist_ok=True)
(db_dir / "SKILL.md").write_text(
"""---
id: "implementation.sqlite_builder"
name: "SQLite Schema & Migration Builder"
version: "1.0.0"
stage: "implementation"
description: "Creates relational tables, DDL migrations, and repository access layers."
tags: ["database", "sql", "sqlite"]
contracts:
input_schema: "DatabaseSpec"
output_schema: "SQLSchemaArtifacts"
---
Generate clean DDL scripts with strict primary and foreign key constraints.
""",
encoding="utf-8",
)
if __name__ == "__main__":
mock_dir = Path("./mock_skills_library")
setup_mock_skill_library(mock_dir)
# Scan and index library
registry = SkillLibraryRegistry(mock_dir)
registry.scan()
# Instantiate orchestrator
orchestrator = Stage1PlanningOrchestrator(registry=registry)
sample_problem = (
"Build a local caching service that watches a folder for incoming CSV telemetry files, "
"validates their headers against a schema, and loads clean records into an SQLite DB."
)
plan_output = orchestrator.run(sample_problem)
print("\n" + "=" * 60)
print("STAGE 1 PLANNING OUTPUT")
print("=" * 60)
print(f"CORE OBJECTIVE: {plan_output.refined_problem.core_objective}\n")
print(f"CONSTRAINTS: {plan_output.refined_problem.identified_constraints}\n")
print("DECOMPOSED EXECUTION PLAN:")
for task in plan_output.execution_plan:
print(f" - [{task.task_id}] {task.title}")
print(f" Skills Required : {task.recommended_skills}")
print(f" Artifact Target : {task.expected_artifact}")
print(f" Dependencies : {task.dependencies}")
Does this include the initial prompt restate and refine step?
Yes, but only in a single combined step within the previous implementation.
In that version, the LLM received both tasks at once in a single prompt call: it extracted the RefinedProblem fields (core_objective, scope_in, scope_out, identified_constraints) and generated the execution_plan simultaneously in the Stage1PlanningOutput JSON response.
Why Separating Them Matters
For high-reliability workflows, bundling refinement and decomposition into one call introduces two risks:
Compounded Assumptions: If the model misunderstands an ambiguous constraint during step 1, it immediately bakes that flaw into the downstream tasks without a verification checkpoint.
Loss of Dedicated Skill Context: A true dedicated Refiner Agent should focus purely on requirements engineering, edge-case analysis, and ambiguity isolation without being distracted by task decomposition or tool binding.
Decoupled 2-Step Architecture
To make the refinement an explicit, deterministic phase before planning begins, split Stage 1 into two dedicated agent interactions:
[Raw User Statement]
│
▼
[Agent 1: Problem Refiner] ◄── Hydrated with `planning.problem_refiner` SKILL.md
│
▼ (Produces structured RefinedProblem contract)
[Checkpoint / Optional Human Review / Validation]
│
▼
[Agent 2: Task Decomposer] ◄── Ingests RefinedProblem + Skill Library Manifests
│
▼
[Structured Execution Plan (DAG)] ──► Handed off to Stage 2
Updated Code: Decoupled Stage 1 Pipeline
Here is how Stage1PlanningOrchestrator refactors to run Step 1 (Restate & Refine) and Step 2 (Skill Binding & Decomposition) sequentially:
"""stage1_decoupled_orchestrator.py
Executes an explicit, two-pass Stage 1 pipeline:
Pass 1: Restate & Refine the problem statement via a specialized Refiner Agent.
Pass 2: Decompose the refined specification into planned tasks mapped to skill manifests.
"""
from __future__ import annotations
import json
import logging
from typing import List, Optional
from google import genai
from google.genai import types
from schemas import (
PlannedTask,
RefinedProblem,
SkillLibraryRegistry,
Stage1PlanningOutput,
)
logger = logging.getLogger("PlanningOrchestrator")
class Stage1PlanningOrchestrator:
def __init__(self, registry: SkillLibraryRegistry, model_id: str = "gemini-2.5-pro"):
self.registry = registry
self.model_id = model_id
self.client = genai.Client()
def refine_problem_statement(self, raw_user_prompt: str) -> RefinedProblem:
"""Step 1: Ingests raw user prompt, extracts assumptions, scope, and formal constraints."""
logger.info("Step 1: Initiating Problem Refinement & Disambiguation...")
refiner_instructions = self.registry.get_instructions("planning.problem_refiner")
system_prompt = (
"You are an expert Systems Requirements & Disambiguation Agent.\n"
"Your sole objective is to inspect raw, ambiguous user problem statements, "
"restate the core objective with high precision, isolate scope boundaries, "
"and identify unspoken technical assumptions and failure modes.\n\n"
f"### Domain Directives:\n{refiner_instructions}"
)
response = self.client.models.generate_content(
model=self.model_id,
contents=[types.Content(role="user", parts=[types.Part.from_text(raw_user_prompt)])],
config=types.GenerateContentConfig(
system_instruction=system_prompt,
response_mime_type="application/json",
response_schema=RefinedProblem,
temperature=0.0, # Zero temperature for deterministic extraction
),
)
refined: RefinedProblem = RefinedProblem.model_validate_json(response.text)
logger.info("Step 1 complete. Core objective: %s", refined.core_objective)
return refined
def decompose_and_bind_skills(self, refined_problem: RefinedProblem) -> Stage1PlanningOutput:
"""Step 2: Takes the refined problem specification and decomposes it into tasks mapped to library skills."""
logger.info("Step 2: Decomposing refined spec and binding available skills...")
skill_manifests = json.dumps(self.registry.list_manifests(), indent=2)
system_prompt = f"""You are a Systems Architect & Task Decomposition Agent.
You take a formally refined problem specification and break it down into an actionable,
sequential execution plan (DAG).
### Available Skill Catalog (Metadata Only):
{skill_manifests}
### Instructions:
1. Decompose the implementation into sequential, non-redundant PlannedTasks.
2. For each task's `recommended_skills`, match the exact `id` strings from the Available Skill Catalog.
3. Establish clear predecessor task dependencies (`dependencies`).
4. Detail the expected concrete artifact for each task.
"""
user_content = f"Formal Problem Specification:\n{refined_problem.model_dump_json(indent=2)}"
class DecompositionPlan(RefinedProblem):
pass # Helper wrapper if needed, or target the final container directly
class FinalPlanSchema(Stage1PlanningOutput):
pass
response = self.client.models.generate_content(
model=self.model_id,
contents=[types.Content(role="user", parts=[types.Part.from_text(user_content)])],
config=types.GenerateContentConfig(
system_instruction=system_prompt,
response_mime_type="application/json",
response_schema=FinalPlanSchema,
temperature=0.1,
),
)
plan: Stage1PlanningOutput = Stage1PlanningOutput.model_validate_json(response.text)
# Guarantee the verified refined problem from Step 1 is carried through verbatim
plan.refined_problem = refined_problem
logger.info("Step 2 complete. Generated %d planned tasks.", len(plan.execution_plan))
return plan
def run(self, raw_user_prompt: str) -> Stage1PlanningOutput:
"""Full pipeline runner executing Step 1 followed by Step 2."""
# 1. Restate & Refine
refined_spec = self.refine_problem_statement(raw_user_prompt)
# 2. Plan, Decompose & Bind Skills
complete_plan = self.decompose_and_bind_skills(refined_spec)
return complete_plan
Implement an Evaluator-Critic verification loop that scores the RefinedProblem schema for completeness before allowing it to proceed to task decomposition.
An Evaluator-Critic verification loop ensures that the RefinedProblem specification is mathematically and semantically complete prior to downstream decomposition. It combines deterministic pre-validation checks (catching empty arrays, vague placeholders, and zero-length boundaries) with a critic agent evaluating clarity, testability, and constraint coverage against an explicit score threshold.
[Raw Problem Statement]
│
▼
┌──────────────────► [Refiner Agent]
│ │
│ ▼
│ [RefinedProblem]
│ │
│ ▼
│ [1. Deterministic Rule Engine] ──(Fails rules)──┐
│ │ │
│ ▼ (Passes rules) │
│ [2. Evaluator-Critic Agent] │
│ │ │
│ ┌───────────┴───────────┐ │
│ Score < Threshold Score >= Threshold │
│ │ │ │
└──(Feedback)───┘ ▼ │
[Passed to Task Decomposer] │
▲ │
└─(Retry Feedback)──┘
Verification Schemas (verification_schemas.py)
"""verification_schemas.py
Pydantic contracts for deterministic audits and the Evaluator-Critic loop.
"""
from __future__ import annotations
from typing import List
from pydantic import BaseModel, Field, conint
class RubricEvaluation(BaseModel):
category: str = Field(..., description="E.g., Ambiguity Elimination, Boundary Definition, Feasibility")
score: conint(ge=1, le=10) = Field(..., description="1-10 rating for this dimension")
findings: str = Field(..., description="Direct critique, identifying unaddressed risks or gaps")
class CriticEvaluationResult(BaseModel):
overall_score: conint(ge=1, le=10) = Field(
..., description="Overall calculated rating across dimensions"
)
passed: bool = Field(
..., description="True if overall_score >= threshold and no blocking issues exist"
)
rubric_breakdown: List[RubricEvaluation] = Field(default_factory=list)
actionable_feedback: List[str] = Field(
default_factory=list,
description="Explicit instructions the Refiner Agent must apply in the next revision iteration",
)
Verification Engine & Refiner Loop (verified_refiner.py)
This module executes:
Fast, deterministic AST/rule checks (catching boilerplate like "N/A", "TBD", or empty fields).
The LLM Critic evaluation with structured Pydantic scoring.
The iterative refinement loop with max-retry bounding and historical feedback accumulation.
"""verified_refiner.py
Orchestrates the Refiner Agent through an Evaluator-Critic feedback loop.
"""
from __future__ import annotations
import json
import logging
from typing import List, Optional, Tuple
from google import genai
from google.genai import types
from schemas import RefinedProblem
from verification_schemas import CriticEvaluationResult
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("VerifiedRefiner")
class DeterministicProblemValidator:
"""Zero-cost heuristic rule engine to catch basic formatting and completeness issues."""
FORBIDDEN_PLACEHOLDERS = {"tbd", "n/a", "none", "todo", "unknown", "placeholder"}
@classmethod
def validate(cls, spec: RefinedProblem) -> Tuple[bool, List[str]]:
violations: List[str] = []
if len(spec.core_objective.strip().split()) < 6:
violations.append("`core_objective` is too terse. It must thoroughly specify the primary goal.")
if not spec.scope_in:
violations.append("`scope_in` must list at least one concrete in-scope deliverable.")
if not spec.scope_out:
violations.append("`scope_out` must explicitly bound the system (negative constraints).")
if not spec.identified_constraints:
violations.append("`identified_constraints` cannot be empty. Detail operational/runtime limits.")
for field_name, items in [
("scope_in", spec.scope_in),
("scope_out", spec.scope_out),
("identified_constraints", spec.identified_constraints),
("key_assumptions", spec.key_assumptions),
]:
for item in items:
if item.strip().lower() in cls.FORBIDDEN_PLACEHOLDERS:
violations.append(f"Forbidden placeholder found in `{field_name}`: '{item}'.")
return (len(violations) == 0, violations)
class EvaluatorCriticLoop:
"""Manages iterative problem refinement with critic evaluation and guardrails."""
def __init__(
self,
model_id: str = "gemini-2.5-pro",
score_threshold: int = 8,
max_iterations: int = 3,
):
self.client = genai.Client()
self.model_id = model_id
self.score_threshold = score_threshold
self.max_iterations = max_iterations
def run_critic(
self, raw_prompt: str, candidate_spec: RefinedProblem
) -> CriticEvaluationResult:
"""Evaluator Agent: Grades candidate problem specification on clarity and completeness."""
system_instruction = f"""You are a Lead Systems Safety & Requirements Verification Auditor.
Your sole job is to ruthlessly critique a `RefinedProblem` candidate against the user's original raw request.
### Evaluation Rubric:
1. **Ambiguity Elimination**: Has every vague buzzword or ambiguous acronym been nailed down to measurable targets?
2. **Negative Scoping (Boundaries)**: Does `scope_out` clearly identify non-goals and adjacent systems to prevent scope creep?
3. **Constraint Feasibility**: Are environmental, operational, runtime, and platform constraints realistic and explicit?
4. **Assumption Soundness**: Are hidden assumptions about dependencies, networks, storage, or permissions uncovered?
### Grading Standard:
- Passing threshold: {self.score_threshold}/10 overall.
- If any critical operational dimension is missing, fail the candidate (passed=False) and give concrete rewrite directives in `actionable_feedback`.
"""
user_content = f"""Original User Prompt:
{raw_prompt}
Candidate Refined Problem:
{candidate_spec.model_dump_json(indent=2)}
"""
response = self.client.models.generate_content(
model=self.model_id,
contents=[types.Content(role="user", parts=[types.Part.from_text(user_content)])],
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=CriticEvaluationResult,
temperature=0.0,
),
)
return CriticEvaluationResult.model_validate_json(response.text)
def execute_refinement_iteration(
self,
raw_prompt: str,
refiner_instructions: str,
feedback_history: List[str],
prior_spec: Optional[RefinedProblem] = None,
) -> RefinedProblem:
"""Refiner Agent: Generates or revises the RefinedProblem using feedback from previous attempts."""
feedback_block = ""
if feedback_history:
formatted_feedback = "\n".join(f"- {f}" for f in feedback_history)
feedback_block = f"\n\n### CRITIQUE AUDIT FEEDBACK FROM PRIOR REVISION:\n{formatted_feedback}\nDirectly resolve each finding above."
prior_draft_block = ""
if prior_spec:
prior_draft_block = f"\n\n### PREVIOUS REFINED DRAFT:\n{prior_spec.model_dump_json(indent=2)}"
system_instruction = f"""You are an expert Systems Requirements & Disambiguation Engineer.
Ingest the user prompt and generate an unambiguous, highly structured RefinedProblem specification.
{refiner_instructions}
{feedback_block}
{prior_draft_block}
"""
response = self.client.models.generate_content(
model=self.model_id,
contents=[types.Content(role="user", parts=[types.Part.from_text(raw_prompt)])],
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=RefinedProblem,
temperature=0.1,
),
)
return RefinedProblem.model_validate_json(response.text)
def refine_with_verification(
self, raw_prompt: str, refiner_instructions: str = ""
) -> Tuple[RefinedProblem, CriticEvaluationResult]:
"""Runs the loop until passing criteria are met or max_iterations is reached."""
feedback_history: List[str] = []
candidate_spec: Optional[RefinedProblem] = None
last_evaluation: Optional[CriticEvaluationResult] = None
for iteration in range(1, self.max_iterations + 1):
logger.info("Starting Refinement Loop Iteration %d/%d", iteration, self.max_iterations)
# 1. Produce / update candidate specification
candidate_spec = self.execute_refinement_iteration(
raw_prompt=raw_prompt,
refiner_instructions=refiner_instructions,
feedback_history=feedback_history,
prior_spec=candidate_spec,
)
# 2. Deterministic rule check
passed_rules, violations = DeterministicProblemValidator.validate(candidate_spec)
if not passed_rules:
logger.warning(
"Deterministic validation failed on iteration %d: %s",
iteration,
violations,
)
feedback_history.extend(violations)
continue
# 3. Critic Agent evaluation
evaluation = self.run_critic(raw_prompt, candidate_spec)
last_evaluation = evaluation
logger.info(
"Critic scored spec: %d/10 (Passed: %s)",
evaluation.overall_score,
evaluation.passed,
)
# Check convergence
if evaluation.passed and evaluation.overall_score >= self.score_threshold:
logger.info("Refinement approved by Critic on iteration %d.", iteration)
return candidate_spec, evaluation
# Accumulate feedback for subsequent iteration
logger.info("Spec rejected. Incorporating %d feedback points.", len(evaluation.actionable_feedback))
feedback_history.extend(evaluation.actionable_feedback)
logger.warning(
"Max iterations (%d) reached without meeting target threshold (%d). Returning best candidate.",
self.max_iterations,
self.score_threshold,
)
return candidate_spec, last_evaluation
Integration into Stage1PlanningOrchestrator
Wire the loop directly into the planning orchestrator's run() method before dispatching tasks to decomposition:
from verified_refiner import EvaluatorCriticLoop
class Stage1PlanningOrchestrator:
def __init__(self, registry, model_id: str = "gemini-2.5-pro"):
self.registry = registry
self.model_id = model_id
self.verifier = EvaluatorCriticLoop(model_id=self.model_id, score_threshold=8)
def run(self, raw_user_prompt: str) -> Stage1PlanningOutput:
# Step 1: Verified Refinement (Evaluator-Critic Loop)
refiner_instructions = self.registry.get_instructions("planning.problem_refiner")
refined_spec, evaluation = self.verifier.refine_with_verification(
raw_prompt=raw_user_prompt,
refiner_instructions=refiner_instructions,
)
logger.info(
"Verified specification produced (Audit Score: %d/10). Proceeding to Step 2...",
evaluation.overall_score,
)
# Step 2: Task Decomposition & Skill Binding
complete_plan = self.decompose_and_bind_skills(refined_spec)
complete_plan.audit_notes += f" | Verified via Critic Loop (Score: {evaluation.overall_score}/10)"
return complete_planImplement an automated circuit breaker and fallback strategy when the Evaluator-Critic loop fails to reach the score threshold within max iterations.
An automated circuit breaker isolates iterative divergence or model stagnation, preventing infinite inference loops, degraded downstream task decomposition, and silent failures. When the verification loop fails to converge above the required score threshold within max_iterations, the circuit trips into one of three deterministic escalation states:
STAGNATION_DETECTED: The critic score fails to improve (or regresses) across successive iterations ($\Delta \text{score} \le 0$).
OSCILLATION_DETECTED: The refiner alternates between contradictory revisions without resolving underlying findings.
THRESHOLD_EXHAUSTION: Iteration limit reached without reaching the passing score threshold.
The fallback engine responds through a tiered hierarchy:
Level 1 (Deterministic Patch & Relax): Programmatically inject missing boundary elements into the best candidate and check if relaxed safety bounds pass.
Level 2 (Model Escalation): Re-hydrate the critique history into a higher-tier reasoning model (e.g., escalating from a flash model to a pro model) for a one-shot reconciliation pass.
Level 3 (Safe Partial Quarantine / Human Handoff): Quarantine the unverified fields, flag missing constraints in an audit envelope, and halt or route to an operator breakpoint.
Circuit Breaker & Fallback Architecture
Plaintext
[Iteration Loop: Refiner ◄──► Critic]
│
▼
[Circuit State Monitor]
├── Delta Tracking (Δ Score)
└── Iteration Budget Check
│
┌────────────────┴────────────────┐
[Score >= Min] [Trip Detected]
│ │
▼ ▼
[Proceed] [Circuit Breaker Trips]
│
┌───────────────────────┼───────────────────────┐
▼ ▼ ▼
[Fallback L1] [Fallback L2] [Fallback L3]
Deterministic Patch Model Escalation Pass Quarantine & Handoff
Schema Definitions (circuit_schemas.py)
"""circuit_schemas.py
Pydantic contracts for Circuit Breaker states, fallback envelopes, and telemetry.
"""
from __future__ import annotations
from enum import Enum
from typing import List, Optional, Dict, Any
from pydantic import BaseModel, Field
from schemas import RefinedProblem
from verification_schemas import CriticEvaluationResult
class CircuitBreakerStatus(str, Enum):
CLOSED = "closed" # Normal operation; loop running or converged
TRIPPED = "tripped" # Failure threshold hit; fallback activated
RECOVERED = "recovered" # Fallback strategy produced a compliant spec
QUARANTINED = "quarantined" # Fallback exhausted; human/supervisor escalation needed
class TripReason(str, Enum):
NONE = "none"
THRESHOLD_EXHAUSTED = "threshold_exhausted"
STAGNATION = "stagnation_detected"
OSCILLATION = "oscillation_detected"
DETERMINISTIC_REJECT = "persistent_deterministic_violations"
class FallbackStrategy(str, Enum):
NONE = "none"
DETERMINISTIC_PATCH = "deterministic_patch"
MODEL_ESCALATION = "model_escalation"
SAFE_DEFAULT_QUARANTINE = "safe_default_quarantine"
class CircuitBreakerEnvelope(BaseModel):
status: CircuitBreakerStatus
trip_reason: TripReason
fallback_applied: FallbackStrategy
final_score: int = Field(..., ge=0, le=10)
iterations_run: int
score_trajectory: List[int] = Field(default_factory=list)
refined_problem: RefinedProblem
unresolved_critiques: List[str] = Field(default_factory=list)
audit_telemetry: Dict[str, Any] = Field(default_factory=dict)
Circuit Breaker Implementation (circuit_breaker.py)
"""circuit_breaker.py
Implements the Evaluator-Critic Circuit Breaker with deterministic state tracking,
delta stagnation monitoring, and multi-tier fallbacks.
"""
from __future__ import annotations
import copy
import logging
from typing import List, Optional, Tuple
from google import genai
from google.genai import types
from schemas import RefinedProblem
from verification_schemas import CriticEvaluationResult
from circuit_schemas import (
CircuitBreakerEnvelope,
CircuitBreakerStatus,
FallbackStrategy,
TripReason,
)
from verified_refiner import DeterministicProblemValidator, EvaluatorCriticLoop
logger = logging.getLogger("CircuitBreaker")
class CircuitBreakerEvaluatorCritic:
"""Wraps the verification loop with trip-wire telemetry and fallback recovery."""
def __init__(
self,
base_model_id: str = "gemini-2.5-flash",
escalation_model_id: str = "gemini-2.5-pro",
score_threshold: int = 8,
min_acceptable_threshold: int = 6,
max_iterations: int = 3,
stagnation_tolerance: int = 1, # Max allowable consecutive non-improving iterations
):
self.base_model_id = base_model_id
self.escalation_model_id = escalation_model_id
self.score_threshold = score_threshold
self.min_acceptable_threshold = min_acceptable_threshold
self.max_iterations = max_iterations
self.stagnation_tolerance = stagnation_tolerance
self.client = genai.Client()
self.verifier = EvaluatorCriticLoop(
model_id=self.base_model_id,
score_threshold=self.score_threshold,
max_iterations=1, # Driven iteratively by this controller
)
def _apply_deterministic_patch(
self, spec: RefinedProblem, violations: List[str]
) -> RefinedProblem:
"""Fallback L1: Programmatically injects missing standard boundaries and scrubs placeholders."""
patched = copy.deepcopy(spec)
logger.info("Executing Fallback L1: Deterministic boundary synthesis...")
if not patched.scope_out:
patched.scope_out = [
"External third-party service modifications outside defined local API contracts",
"Unspecified persistent hardware/infrastructure provisioning",
]
if not patched.identified_constraints:
patched.identified_constraints = [
"Local environment runtime execution only",
"Strict validation against predefined local Pydantic schemas",
]
# Clean forbidden placeholder tokens
scrubbed_in = [
item for item in patched.scope_in
if item.lower() not in DeterministicProblemValidator.FORBIDDEN_PLACEHOLDERS
]
patched.scope_in = scrubbed_in or [patched.core_objective]
return patched
def _execute_model_escalation(
self, raw_prompt: str, candidate_spec: RefinedProblem, critique_history: List[str]
) -> RefinedProblem:
"""Fallback L2: Elevates reasoning context to an advanced frontier model for one-shot synthesis."""
logger.info(
"Executing Fallback L2: Escalating to high-tier reasoning model [%s]...",
self.escalation_model_id,
)
system_instruction = (
"You are an Executive Principal Systems Architect acting as an emergency reconciler.\n"
"A lower-tier agent failed to resolve specification ambiguities across iterative loops.\n"
"Your task is to ingest the raw user statement, the failed draft, and all accumulated auditor critiques, "
"then produce a fully reconciled, definitive, and highly constrained RefinedProblem specification.\n"
"Do not leave any placeholder or unaddressed boundary."
)
user_content = f"""Raw Problem Statement:
{raw_prompt}
Failed Candidate Draft:
{candidate_spec.model_dump_json(indent=2)}
Accumulated Auditor Findings:
{chr(10).join(f"- {c}" for c in critique_history)}
"""
response = self.client.models.generate_content(
model=self.escalation_model_id,
contents=[types.Content(role="user", parts=[types.Part.from_text(user_content)])],
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=RefinedProblem,
temperature=0.0,
),
)
return RefinedProblem.model_validate_json(response.text)
def run(self, raw_prompt: str, refiner_instructions: str = "") -> CircuitBreakerEnvelope:
"""Executes the loop, tracks state metrics, and intercepts failure conditions."""
score_trajectory: List[int] = []
critique_history: List[str] = []
best_candidate: Optional[RefinedProblem] = None
best_score = -1
last_eval: Optional[CriticEvaluationResult] = None
stagnation_counter = 0
for iteration in range(1, self.max_iterations + 1):
logger.info("Circuit Breaker [Status: CLOSED] - Starting Iteration %d/%d", iteration, self.max_iterations)
# 1. Generate / Refine Candidate
candidate = self.verifier.execute_refinement_iteration(
raw_prompt=raw_prompt,
refiner_instructions=refiner_instructions,
feedback_history=critique_history,
prior_spec=best_candidate,
)
# 2. Rule Check
valid_rules, rule_violations = DeterministicProblemValidator.validate(candidate)
if not valid_rules:
logger.warning("Deterministic validation violations: %s", rule_violations)
critique_history.extend(rule_violations)
current_score = 1
else:
# 3. Critic Scoring
evaluation = self.verifier.run_critic(raw_prompt, candidate)
last_eval = evaluation
current_score = evaluation.overall_score
critique_history.extend(evaluation.actionable_feedback)
score_trajectory.append(current_score)
# Update high-water mark candidate
if current_score > best_score:
best_score = current_score
best_candidate = candidate
# Check convergence
if current_score >= self.score_threshold and valid_rules:
logger.info("Specification converged normally at iteration %d (Score: %d).", iteration, current_score)
return CircuitBreakerEnvelope(
status=CircuitBreakerStatus.CLOSED,
trip_reason=TripReason.NONE,
fallback_applied=FallbackStrategy.NONE,
final_score=current_score,
iterations_run=iteration,
score_trajectory=score_trajectory,
refined_problem=candidate,
unresolved_critiques=[],
audit_telemetry={"status": "Optimal Convergence"},
)
# Stagnation monitoring
if len(score_trajectory) > 1:
delta = score_trajectory[-1] - score_trajectory[-2]
if delta <= 0:
stagnation_counter += 1
else:
stagnation_counter = 0
# Trip Condition A: Stagnation threshold exceeded
if stagnation_counter >= self.stagnation_tolerance:
logger.error("Circuit Breaker TRIPPED: Stagnation detected across %d iterations.", stagnation_counter)
return self._trigger_fallbacks(
raw_prompt=raw_prompt,
best_candidate=best_candidate or candidate,
best_score=best_score,
score_trajectory=score_trajectory,
critiques=critique_history,
iterations=iteration,
reason=TripReason.STAGNATION,
)
# Trip Condition B: Iteration limit exhausted without reaching target threshold
logger.error("Circuit Breaker TRIPPED: Iteration limit exhausted without convergence.")
return self._trigger_fallbacks(
raw_prompt=raw_prompt,
best_candidate=best_candidate,
best_score=best_score,
score_trajectory=score_trajectory,
critiques=critique_history,
iterations=self.max_iterations,
reason=TripReason.THRESHOLD_EXHAUSTED,
)
def _trigger_fallbacks(
self,
raw_prompt: str,
best_candidate: RefinedProblem,
best_score: int,
score_trajectory: List[int],
critiques: List[str],
iterations: int,
reason: TripReason,
) -> CircuitBreakerEnvelope:
"""Executes the hierarchical escalation ladder: L1 -> L2 -> L3."""
logger.warning("Initiating Tiered Fallback Engine for reason: %s", reason.value)
# Fallback L1: Deterministic patching
patched_spec = self._apply_deterministic_patch(best_candidate, critiques)
l1_eval = self.verifier.run_critic(raw_prompt, patched_spec)
if l1_eval.overall_score >= self.min_acceptable_threshold:
logger.info("Fallback L1 Succeeded: Critic score recovered to %d/10.", l1_eval.overall_score)
return CircuitBreakerEnvelope(
status=CircuitBreakerStatus.RECOVERED,
trip_reason=reason,
fallback_applied=FallbackStrategy.DETERMINISTIC_PATCH,
final_score=l1_eval.overall_score,
iterations_run=iterations,
score_trajectory=score_trajectory + [l1_eval.overall_score],
refined_problem=patched_spec,
unresolved_critiques=l1_eval.actionable_feedback,
audit_telemetry={"recovery_tier": "L1_Deterministic_Patch"},
)
# Fallback L2: Model escalation
escalated_spec = self._execute_model_escalation(raw_prompt, patched_spec, critiques)
l2_eval = self.verifier.run_critic(raw_prompt, escalated_spec)
if l2_eval.overall_score >= self.min_acceptable_threshold:
logger.info("Fallback L2 Succeeded: Model escalation reached %d/10.", l2_eval.overall_score)
return CircuitBreakerEnvelope(
status=CircuitBreakerStatus.RECOVERED,
trip_reason=reason,
fallback_applied=FallbackStrategy.MODEL_ESCALATION,
final_score=l2_eval.overall_score,
iterations_run=iterations,
score_trajectory=score_trajectory + [l2_eval.overall_score],
refined_problem=escalated_spec,
unresolved_critiques=l2_eval.actionable_feedback,
audit_telemetry={"recovery_tier": "L2_Model_Escalation"},
)
# Fallback L3: Quarantine & Safe Partitioning
logger.critical("All automatic fallbacks exhausted. Routing to Quarantine Envelope.")
return CircuitBreakerEnvelope(
status=CircuitBreakerStatus.QUARANTINED,
trip_reason=reason,
fallback_applied=FallbackStrategy.SAFE_DEFAULT_QUARANTINE,
final_score=max(best_score, l2_eval.overall_score),
iterations_run=iterations,
score_trajectory=score_trajectory + [l2_eval.overall_score],
refined_problem=escalated_spec,
unresolved_critiques=list(set(critiques + l2_eval.actionable_feedback)),
audit_telemetry={
"alert": "OPERATOR_INTERVENTION_REQUIRED",
"recommended_action": "Inspect unresolved_critiques before proceeding to Stage 1 decomposition.",
},
)
Integration into Stage 1 Planning Pipeline
"""stage1_orchestrator_integration.py
Demonstrates how the Planning Orchestrator evaluates the CircuitBreakerEnvelope.
"""
from circuit_breaker import CircuitBreakerEvaluatorCritic
from circuit_schemas import CircuitBreakerStatus
class Stage1PlanningOrchestrator:
def __init__(self, registry):
self.registry = registry
self.circuit_breaker = CircuitBreakerEvaluatorCritic(
base_model_id="gemini-2.5-flash",
escalation_model_id="gemini-2.5-pro",
score_threshold=8,
min_acceptable_threshold=6,
max_iterations=3,
stagnation_tolerance=1,
)
def run(self, raw_user_prompt: str):
refiner_instructions = self.registry.get_instructions("planning.problem_refiner")
# Execute verified refinement under circuit breaker supervision
envelope = self.circuit_breaker.run(raw_user_prompt, refiner_instructions)
if envelope.status == CircuitBreakerStatus.QUARANTINED:
# Raise exception, write audit dump, or trigger interactive CLI prompt
raise SystemError(
f"Stage 1 Refinement Halted: Specification quarantined after {envelope.iterations_run} iterations. "
f"Trip Reason: {envelope.trip_reason.value}. Unresolved: {envelope.unresolved_critiques}"
)
logger.info(
"Specification validated for Stage 1 decomposition (Status: %s | Score: %d | Strategy: %s)",
envelope.status.value,
envelope.final_score,
envelope.fallback_applied.value,
)
# Proceed to step 2 (Task Decomposition) using envelope.refined_problem
return self.decompose_and_bind_skills(envelope.refined_problem)Implement an audit logging utility that writes the CircuitBreakerEnvelope, trajectory curves, and model transitions to a JSON audit file.
Audit Logging Architecture
An audit log for high-reliability agent workflows must capture four distinct dimensions of runtime execution:
The Circuit Breaker Envelope: Final status, trip reason, recovery strategy, and the resulting
RefinedProblem.Iteration Trajectory & Deltas: Step-by-step score progression, deterministic rule violation records, and critic feedback per iteration.
Model & Tier Transitions: Clear telemetry recording which models were invoked (e.g., base model to escalation model) and tokens/latency metrics.
Deterministic Reproducibility Metadata: Session UUIDs, ISO-8601 UTC timestamps, schema versions, and runtime parameters.
Audit Log Schema (audit_schemas.py)
"""audit_schemas.py
Pydantic contracts for structured agent session auditing and telemetry.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
from circuit_schemas import CircuitBreakerEnvelope
class StepTelemetry(BaseModel):
iteration: int
timestamp_utc: str = Field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
actor_model_id: str
critic_model_id: str
deterministic_passed: bool
deterministic_violations: List[str] = Field(default_factory=list)
critic_score: Optional[int] = Field(None, ge=0, le=10)
score_delta_from_prior: Optional[int] = None
actionable_feedback: List[str] = Field(default_factory=list)
latency_ms: Optional[float] = None
class ModelTransitionEvent(BaseModel):
timestamp_utc: str = Field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
transition_type: str = Field(
..., description="E.g., BASE_TO_ESCALATION, DETERMINISTIC_FALLBACK, LOOP_RETRY"
)
from_tier: str
to_tier: str
reason: str
class ProblemRefinementAuditRecord(BaseModel):
audit_id: str = Field(default_factory=lambda: f"audit-{uuid.uuid4().hex[:12]}")
session_id: str
created_at_utc: str = Field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
raw_user_prompt: str
envelope: CircuitBreakerEnvelope
trajectory: List[StepTelemetry] = Field(default_factory=list)
model_transitions: List[ModelTransitionEvent] = Field(default_factory=list)
execution_environment: Dict[str, Any] = Field(default_factory=dict)
Audit Logger Implementation (audit_logger.py)
This utility provides atomic writes to disk using a unique file naming convention (audit_{timestamp}_{session_id}.json), computes real-time score trajectory deltas, and serializes the complete state without losing schema guarantees.
"""audit_logger.py
Manages telemetry aggregation and atomic JSON export for Evaluator-Critic audit logs.
"""
from __future__ import annotations
import json
import logging
import os
import platform
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from audit_schemas import (
ModelTransitionEvent,
ProblemRefinementAuditRecord,
StepTelemetry,
)
from circuit_schemas import CircuitBreakerEnvelope
logger = logging.getLogger("AuditLogger")
class AuditLogger:
"""Thread-safe, non-destructive audit trail logger for agent orchestration."""
def __init__(self, log_dir: str | Path = "./audit_logs", session_id: Optional[str] = None):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(parents=True, exist_ok=True)
self.session_id = session_id or datetime.now(timezone.utc).strftime("sess_%Y%m%d_%H%M%S")
self.trajectory: List[StepTelemetry] = []
self.transitions: List[ModelTransitionEvent] = []
self._last_score: Optional[int] = None
def record_step(
self,
iteration: int,
actor_model_id: str,
critic_model_id: str,
deterministic_passed: bool,
deterministic_violations: List[str],
critic_score: Optional[int],
actionable_feedback: List[str],
latency_ms: Optional[float] = None,
) -> None:
"""Records a single iteration evaluation with auto-computed score trajectory deltas."""
score_delta = None
if critic_score is not None:
if self._last_score is not None:
score_delta = critic_score - self._last_score
self._last_score = critic_score
entry = StepTelemetry(
iteration=iteration,
actor_model_id=actor_model_id,
critic_model_id=critic_model_id,
deterministic_passed=deterministic_passed,
deterministic_violations=deterministic_violations,
critic_score=critic_score,
score_delta_from_prior=score_delta,
actionable_feedback=actionable_feedback,
latency_ms=latency_ms,
)
self.trajectory.append(entry)
logger.debug("Logged audit step iteration %d (Score: %s, Delta: %s)", iteration, critic_score, score_delta)
def record_transition(
self,
transition_type: str,
from_tier: str,
to_tier: str,
reason: str,
) -> None:
"""Records runtime switches between models, fallbacks, and escalation tiers."""
event = ModelTransitionEvent(
transition_type=transition_type,
from_tier=from_tier,
to_tier=to_tier,
reason=reason,
)
self.transitions.append(event)
logger.info("Model transition recorded: %s -> %s (Reason: %s)", from_tier, to_tier, reason)
def export(
self,
raw_user_prompt: str,
envelope: CircuitBreakerEnvelope,
extra_env_metadata: Optional[Dict[str, Any]] = None,
) -> Path:
"""Assembles the final audit record and atomically writes it to disk."""
env_meta = {
"python_version": sys.version.split()[0],
"os_platform": platform.platform(),
"pid": os.getpid(),
}
if extra_env_metadata:
env_meta.update(extra_env_metadata)
record = ProblemRefinementAuditRecord(
session_id=self.session_id,
raw_user_prompt=raw_user_prompt,
envelope=envelope,
trajectory=self.trajectory,
model_transitions=self.transitions,
execution_environment=env_meta,
)
timestamp_slug = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
output_filename = f"audit_{timestamp_slug}_{self.session_id}.json"
output_path = self.log_dir / output_filename
temp_path = output_path.with_suffix(".tmp")
# Atomic file write to avoid partial dumps during process interrupts
with open(temp_path, "w", encoding="utf-8") as f:
f.write(record.model_dump_json(indent=2))
temp_path.replace(output_path)
logger.info("Successfully exported structured audit log to %s", output_path.resolve())
return output_path
Instrumented Circuit Breaker Execution
Here is how the CircuitBreakerEvaluatorCritic calls the AuditLogger at each step, tracking execution time, deltas, and model escalation boundaries.
"""circuit_breaker_instrumented.py
Circuit breaker runner integrated with live telemetry and structured audit generation.
"""
from __future__ import annotations
import time
from typing import Optional
from audit_logger import AuditLogger
from circuit_breaker import CircuitBreakerEvaluatorCritic
from circuit_schemas import CircuitBreakerEnvelope, TripReason
class InstrumentedCircuitBreaker(CircuitBreakerEvaluatorCritic):
def __init__(self, *args, audit_dir: str = "./audit_logs", **kwargs):
super().__init__(*args, **kwargs)
self.audit_logger = AuditLogger(log_dir=audit_dir)
def run_with_audit(
self, raw_prompt: str, refiner_instructions: str = ""
) -> tuple[CircuitBreakerEnvelope, str]:
"""Runs the circuit breaker while recording granular step metrics to JSON."""
score_trajectory = []
critique_history = []
best_candidate = None
best_score = -1
stagnation_counter = 0
for iteration in range(1, self.max_iterations + 1):
t_start = time.perf_counter()
# 1. Refiner generation
candidate = self.verifier.execute_refinement_iteration(
raw_prompt=raw_prompt,
refiner_instructions=refiner_instructions,
feedback_history=critique_history,
prior_spec=best_candidate,
)
# 2. Rule check
valid_rules, rule_violations = self.verifier.validator.validate(candidate) if hasattr(self.verifier, 'validator') else (True, [])
# 3. Critic check
critic_score = None
feedback = []
if valid_rules:
eval_result = self.verifier.run_critic(raw_prompt, candidate)
critic_score = eval_result.overall_score
feedback = eval_result.actionable_feedback
critique_history.extend(feedback)
else:
critique_history.extend(rule_violations)
latency_ms = (time.perf_counter() - t_start) * 1000.0
# Record telemetry step
self.audit_logger.record_step(
iteration=iteration,
actor_model_id=self.base_model_id,
critic_model_id=self.base_model_id,
deterministic_passed=valid_rules,
deterministic_violations=rule_violations,
critic_score=critic_score,
actionable_feedback=feedback if valid_rules else rule_violations,
latency_ms=round(latency_ms, 2),
)
# Check convergence & stagnation
if critic_score is not None:
score_trajectory.append(critic_score)
if critic_score > best_score:
best_score = critic_score
best_candidate = candidate
if critic_score >= self.score_threshold and valid_rules:
envelope = self._build_closed_envelope(
candidate, critic_score, iteration, score_trajectory
)
log_file = self.audit_logger.export(raw_prompt, envelope)
return envelope, str(log_file)
if len(score_trajectory) > 1 and (score_trajectory[-1] - score_trajectory[-2] <= 0):
stagnation_counter += 1
else:
stagnation_counter = 0
if stagnation_counter >= self.stagnation_tolerance:
self.audit_logger.record_transition(
transition_type="CIRCUIT_BREAKER_TRIP",
from_tier=self.base_model_id,
to_tier="FALLBACK_LADDER",
reason=f"Stagnation detected across {stagnation_counter} iterations",
)
break
# Fallback Triggered: Track escalation transition
self.audit_logger.record_transition(
transition_type="MODEL_ESCALATION",
from_tier=self.base_model_id,
to_tier=self.escalation_model_id,
reason="Primary evaluation loop failed to converge above score threshold",
)
envelope = self._trigger_fallbacks(
raw_prompt=raw_prompt,
best_candidate=best_candidate or candidate,
best_score=best_score,
score_trajectory=score_trajectory,
critiques=critique_history,
iterations=self.max_iterations,
reason=TripReason.STAGNATION if stagnation_counter >= self.stagnation_tolerance else TripReason.THRESHOLD_EXHAUSTED,
)
log_file = self.audit_logger.export(raw_prompt, envelope)
return envelope, str(log_file)
Sample Exported JSON Output (audit_sample.json)
{
"audit_id": "audit-4f9e11c8a02b",
"session_id": "sess_20261104_120530",
"created_at_utc": "2026-11-04T18:05:35.128491+00:00",
"raw_user_prompt": "Build a CSV ingest watcher that validates schema and loads clean records into SQLite.",
"envelope": {
"status": "recovered",
"trip_reason": "stagnation_detected",
"fallback_applied": "model_escalation",
"final_score": 9,
"iterations_run": 2,
"score_trajectory": [5, 5, 9],
"refined_problem": {
"core_objective": "Construct a local filesystem directory monitor that validates streaming CSV payloads against predefined Pydantic schemas and ingests valid records into SQLite via transactional batching.",
"scope_in": [
"Inotify/filesystem change watcher daemon",
"Schema validation with error quarantine table",
"Transactional SQLite ingestion"
],
"scope_out": [
"External cloud cloud storage syncing (S3, GCS)",
"Distributed message queues"
],
"identified_constraints": [
"Local Python 3.11+ runtime",
"Zero unhandled thread crashes on corrupted CSV files"
],
"key_assumptions": [
"CSV files are encoded in UTF-8 with standard comma delimiters"
]
},
"unresolved_critiques": [],
"audit_telemetry": {
"recovery_tier": "L2_Model_Escalation"
}
},
"trajectory": [
{
"iteration": 1,
"timestamp_utc": "2026-11-04T18:05:31.011290+00:00",
"actor_model_id": "gemini-2.5-flash",
"critic_model_id": "gemini-2.5-flash",
"deterministic_passed": true,
"deterministic_violations": [],
"critic_score": 5,
"score_delta_from_prior": null,
"actionable_feedback": [
"scope_out does not define boundaries for network/cloud replication."
],
"latency_ms": 1120.45
},
{
"iteration": 2,
"timestamp_utc": "2026-11-04T18:05:32.450123+00:00",
"actor_model_id": "gemini-2.5-flash",
"critic_model_id": "gemini-2.5-flash",
"deterministic_passed": true,
"deterministic_violations": [],
"critic_score": 5,
"score_delta_from_prior": 0,
"actionable_feedback": [
"File locking behavior during concurrent writes was not constrained."
],
"latency_ms": 1054.12
}
],
"model_transitions": [
{
"timestamp_utc": "2026-11-04T18:05:33.502011+00:00",
"transition_type": "CIRCUIT_BREAKER_TRIP",
"from_tier": "gemini-2.5-flash",
"to_tier": "FALLBACK_LADDER",
"reason": "Stagnation detected across 1 iterations"
},
{
"timestamp_utc": "2026-11-04T18:05:33.502450+00:00",
"transition_type": "MODEL_ESCALATION",
"from_tier": "gemini-2.5-flash",
"to_tier": "gemini-2.5-pro",
"reason": "Primary evaluation loop failed to converge above score threshold"
}
],
"execution_environment": {
"python_version": "3.11.8",
"os_platform": "Linux-6.6.0-x86_64",
"pid": 28419
}
}Bring it altogether. Provide a combine Stage 1 and Stage 2 application implemented using Ollama supported local LLMs, and saving the activity log and results to markdown file
Here is the complete, self-contained multi-agent application integrating Stage 1 (Refinement, Circuit Breaker, Critic Loop, and Decomposition) with Stage 2 (Task Execution, Dynamic Skill Hydration, and File/Artifact Generation) powered strictly by local Ollama models, with complete activity logging and final delivery exported directly to an audit-ready Markdown report.
End-to-End Architecture
[User Problem Statement]
│
▼
┌───────────────────────────────┐
│ STAGE 1: PLANNING ENGINE │
├───────────────────────────────┤
│ 1. Restate & Refine Problem │
│ 2. Deterministic Rule Engine │
│ 3. Evaluator-Critic Loop │
│ (w/ Circuit Breaker & Trip)│
│ 4. Decompose & Bind Skills │
└───────────────┬───────────────┘
│
[Structured Execution Plan]
│
▼
┌───────────────────────────────┐
│ STAGE 2: EXECUTION ENGINE │
├───────────────────────────────┤
│ For each Planned Task (DAG): │
│ 1. Discover & Hydrate Skill │
│ 2. Load Prompts & Contracts │
│ 3. Execute via Local Ollama │
│ 4. Verify & Write Artifacts │
└───────────────┬───────────────┘
│
▼
[Complete Markdown Audit & Artifact Report]
Project File Structure
orchestrator_project/
├── skills_library/ # Skill directory structure
│ ├── planning/
│ │ └── problem_refiner/
│ │ └── SKILL.md
│ └── implementation/
│ ├── python_scaffolder/
│ │ └── SKILL.md
│ └── sqlite_builder/
│ └── SKILL.md
├── outputs/ # Target workspace for generated artifacts
├── reports/ # Output Markdown execution logs
├── schemas.py # Unified Pydantic contracts
└── main.py # Complete unified pipeline application
1. Data Contracts & Schemas (schemas.py)
Save this file as schemas.py:
"""schemas.py
Pydantic contracts for Stage 1, Stage 2, Circuit Breaker, and Audit Reporting.
"""
from __future__ import annotations
from enum import Enum
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class OrchestrationStage(str, Enum):
PLANNING = "planning"
IMPLEMENTATION = "implementation"
VERIFICATION = "verification"
GENERAL = "general"
class CircuitBreakerStatus(str, Enum):
CLOSED = "closed"
RECOVERED = "recovered"
QUARANTINED = "quarantined"
class SkillMetadata(BaseModel):
id: str
name: str
version: str = "1.0.0"
stage: OrchestrationStage = OrchestrationStage.GENERAL
description: str
tags: List[str] = Field(default_factory=list)
class RefinedProblem(BaseModel):
core_objective: str = Field(..., description="Precision statement of goal")
scope_in: List[str] = Field(default_factory=list)
scope_out: List[str] = Field(default_factory=list)
identified_constraints: List[str] = Field(default_factory=list)
key_assumptions: List[str] = Field(default_factory=list)
class CriticEvaluationResult(BaseModel):
overall_score: int = Field(..., ge=1, le=10)
passed: bool
findings: str
actionable_feedback: List[str] = Field(default_factory=list)
class PlannedTask(BaseModel):
task_id: str = Field(..., description="E.g., TASK-01")
title: str
description: str
target_stage: OrchestrationStage = OrchestrationStage.IMPLEMENTATION
recommended_skills: List[str] = Field(default_factory=list)
dependencies: List[str] = Field(default_factory=list)
expected_artifact_path: str = Field(..., description="Target file path to generate")
class Stage1PlanningOutput(BaseModel):
refined_problem: RefinedProblem
execution_plan: List[PlannedTask]
audit_notes: str
class TaskExecutionResult(BaseModel):
task_id: str
title: str
skill_used: str
artifact_path: str
generated_content: str
success: bool
runtime_sec: float
2. Unified Application Implementation (main.py)
This standalone script builds the mock skill library if missing, spins up Ollama queries using schema constraints via Pydantic model_dump_json(), runs the Evaluator-Critic loop with circuit breaker logic, executes Stage 2 tasks sequentially, writes output files, and compiles a Markdown activity report.
"""main.py
Autonomous Two-Stage Multi-Agent Orchestrator utilizing local Ollama models.
Stage 1: Restate, Refine, Verify (Evaluator-Critic Circuit Breaker), Decompose.
Stage 2: Dynamic Skill Hydration, Code/Artifact Generation, and Markdown Export.
"""
from __future__ import annotations
import json
import logging
import os
import shutil
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import ollama
import yaml
from schemas import (
CircuitBreakerStatus,
CriticEvaluationResult,
OrchestrationStage,
PlannedTask,
RefinedProblem,
SkillMetadata,
Stage1PlanningOutput,
TaskExecutionResult,
)
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("LocalOrchestrator")
# =====================================================================
# 1. Skill Library Registry
# =====================================================================
class SkillLibraryRegistry:
def __init__(self, root_dir: Path):
self.root_dir = root_dir
self.skills: Dict[str, SkillMetadata] = {}
self.instructions: Dict[str, str] = {}
def scan(self) -> None:
self.skills.clear()
self.instructions.clear()
if not self.root_dir.exists():
return
for skill_path in self.root_dir.glob("**/SKILL.md"):
try:
raw_text = skill_path.read_text(encoding="utf-8")
parts = raw_text.split("---", 2)
if len(parts) < 3:
continue
meta_dict = yaml.safe_load(parts[1])
meta = SkillMetadata(**meta_dict)
self.skills[meta.id] = meta
self.instructions[meta.id] = parts[2].strip()
logger.info("Registered skill [%s] from %s", meta.id, skill_path.name)
except Exception as e:
logger.warning("Failed to parse skill at %s: %e", skill_path, e)
def get_manifests_json(self) -> str:
return json.dumps([s.model_dump() for s in self.skills.values()], indent=2)
def get_instructions(self, skill_id: str) -> str:
return self.instructions.get(skill_id, "Execute standard task instructions accurately.")
# =====================================================================
# 2. Local Ollama Client Wrapper with Structured Pydantic Extraction
# =====================================================================
class LocalOllamaClient:
def __init__(self, model_name: str = "llama3.1:latest"):
self.model_name = model_name
def generate_structured(self, system_prompt: str, user_prompt: str, schema_class: type[BaseModel]) -> BaseModel:
"""Enforces schema-compliant JSON extraction through Ollama's format parameter."""
schema_json = json.dumps(schema_class.model_json_schema())
augmented_system = (
f"{system_prompt}\n\n"
f"You MUST reply with valid JSON matching this JSON Schema exactly:\n{schema_json}\n"
"Do not emit explanatory text, introductory remarks, or markdown code fences outside JSON."
)
response = ollama.chat(
model=self.model_name,
messages=[
{"role": "system", "content": augmented_system},
{"role": "user", "content": user_prompt},
],
format="json",
options={"temperature": 0.1},
)
content = response["message"]["content"]
return schema_class.model_validate_json(content)
def generate_text(self, system_prompt: str, user_prompt: str) -> str:
response = ollama.chat(
model=self.model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
options={"temperature": 0.2},
)
return response["message"]["content"]
# =====================================================================
# 3. Stage 1: Planning Orchestrator with Circuit Breaker
# =====================================================================
class Stage1Planner:
def __init__(self, client: LocalOllamaClient, registry: SkillLibraryRegistry, score_threshold: int = 7):
self.client = client
self.registry = registry
self.score_threshold = score_threshold
self.activity_log: List[Dict[str, Any]] = []
def _deterministic_check(self, spec: RefinedProblem) -> Tuple[bool, List[str]]:
violations = []
if len(spec.core_objective.split()) < 5:
violations.append("Core objective is too brief or ambiguous.")
if not spec.scope_in:
violations.append("scope_in must contain at least one item.")
if not spec.scope_out:
violations.append("scope_out cannot be empty (negative boundaries missing).")
if not spec.identified_constraints:
violations.append("identified_constraints cannot be empty.")
return len(violations) == 0, violations
def refine_with_critic_loop(self, raw_prompt: str, max_iterations: int = 3) -> Tuple[RefinedProblem, CircuitBreakerStatus, List[int]]:
feedback_history: List[str] = []
candidate_spec: Optional[RefinedProblem] = None
score_trajectory: List[int] = []
refiner_instructions = self.registry.get_instructions("planning.problem_refiner")
for iteration in range(1, max_iterations + 1):
logger.info("Stage 1: Refinement Iteration %d/%d", iteration, max_iterations)
# Step 1: Refine
refiner_system = (
"You are an expert Systems Requirements & Problem Disambiguation Agent.\n"
"Restate the problem statement, isolate hard boundaries, and unearth hidden assumptions.\n"
f"{refiner_instructions}\n"
)
if feedback_history:
refiner_system += f"\nPrevious Auditor Critiques to Resolve:\n" + "\n".join(f"- {f}" for f in feedback_history)
candidate_spec = self.client.generate_structured(
system_prompt=refiner_system,
user_prompt=raw_prompt,
schema_class=RefinedProblem,
)
# Step 2: Deterministic Rule Check
valid_rules, violations = self._deterministic_check(candidate_spec)
if not valid_rules:
feedback_history.extend(violations)
score_trajectory.append(2)
self.activity_log.append({
"stage": "Stage 1 - Critic",
"iteration": iteration,
"status": "Deterministic Rule Failure",
"violations": violations,
"score": 2,
})
continue
# Step 3: Critic Evaluation
critic_system = (
"You are a Systems Verification Auditor. Grade the candidate RefinedProblem on a 1-10 scale.\n"
"Evaluate scope clarity, completeness of negative constraints, and feasibility.\n"
f"A score of >= {self.score_threshold} means passed=True."
)
critic_user = f"Original Prompt:\n{raw_prompt}\n\nCandidate Spec:\n{candidate_spec.model_dump_json(indent=2)}"
critique: CriticEvaluationResult = self.client.generate_structured(
system_prompt=critic_system,
user_prompt=critic_user,
schema_class=CriticEvaluationResult,
)
score_trajectory.append(critique.overall_score)
self.activity_log.append({
"stage": "Stage 1 - Critic",
"iteration": iteration,
"score": critique.overall_score,
"passed": critique.passed,
"findings": critique.findings,
"feedback": critique.actionable_feedback,
})
if critique.passed and critique.overall_score >= self.score_threshold:
logger.info("Problem Refinement approved on iteration %d with score %d/10", iteration, critique.overall_score)
return candidate_spec, CircuitBreakerStatus.CLOSED, score_trajectory
feedback_history.extend(critique.actionable_feedback)
# Check Stagnation
if len(score_trajectory) > 1 and score_trajectory[-1] <= score_trajectory[-2]:
logger.warning("Circuit Breaker Stagnation detected. Tripping to deterministic recovery.")
candidate_spec.scope_out.append("External cloud infra outside local execution boundary")
candidate_spec.identified_constraints.append("Must run offline using local Python and Ollama")
return candidate_spec, CircuitBreakerStatus.RECOVERED, score_trajectory
return candidate_spec, CircuitBreakerStatus.QUARANTINED, score_trajectory
def decompose(self, refined_spec: RefinedProblem) -> Stage1PlanningOutput:
logger.info("Stage 1: Decomposing refined spec and mapping skill catalog...")
catalog_manifests = self.registry.get_manifests_json()
system_prompt = (
"You are a Master Systems Architect & Task Decomposition Orchestrator.\n"
"Decompose the formal problem specification into sequential execution tasks.\n"
"For each task, assign one or more matching skill IDs from the Available Skill Catalog.\n"
f"Available Skill Catalog:\n{catalog_manifests}"
)
user_prompt = f"Problem Specification:\n{refined_spec.model_dump_json(indent=2)}"
plan: Stage1PlanningOutput = self.client.generate_structured(
system_prompt=system_prompt,
user_prompt=user_prompt,
schema_class=Stage1PlanningOutput,
)
plan.refined_problem = refined_spec
return plan
# =====================================================================
# 4. Stage 2: Execution Orchestrator
# =====================================================================
class Stage2Executor:
def __init__(self, client: LocalOllamaClient, registry: SkillLibraryRegistry, output_workspace: Path):
self.client = client
self.registry = registry
self.output_workspace = output_workspace
self.output_workspace.mkdir(parents=True, exist_ok=True)
self.results: List[TaskExecutionResult] = []
def execute_plan(self, plan: Stage1PlanningOutput) -> List[TaskExecutionResult]:
logger.info("Initiating Stage 2 Execution for %d planned tasks...", len(plan.execution_plan))
for task in plan.execution_plan:
t_start = time.perf_counter()
logger.info("Executing [%s]: %s", task.task_id, task.title)
# 1. Resolve & Hydrate Skill Context
assigned_skill_id = task.recommended_skills[0] if task.recommended_skills else "general"
skill_instructions = self.registry.get_instructions(assigned_skill_id)
# 2. Formulate Agent Persona
agent_system_prompt = (
f"You are a Specialized Execution Agent powered by the [{assigned_skill_id}] skill.\n"
f"Your task is: {task.title}\n"
f"Description: {task.description}\n"
f"Target File: {task.expected_artifact_path}\n\n"
f"### Skill Operational Instructions:\n{skill_instructions}\n\n"
"Produce the complete, working, production-grade file content without truncated sections."
)
agent_user_prompt = (
f"Generate the exact file artifact content for '{task.expected_artifact_path}' to satisfy this task.\n"
f"System Objective: {plan.refined_problem.core_objective}\n"
f"System Constraints: {plan.refined_problem.identified_constraints}"
)
generated_content = self.client.generate_text(agent_system_prompt, agent_user_prompt)
# Clean markdown code fences if wrapped
clean_content = generated_content.strip()
if clean_content.startswith("```"):
lines = clean_content.splitlines()
if len(lines) >= 3 and lines[0].startswith("```") and lines[-1].startswith("```"):
clean_content = "\n".join(lines[1:-1])
# 3. Write Concrete File Artifact
artifact_file = self.output_workspace / task.expected_artifact_path
artifact_file.parent.mkdir(parents=True, exist_ok=True)
artifact_file.write_text(clean_content, encoding="utf-8")
elapsed = round(time.perf_counter() - t_start, 2)
self.results.append(
TaskExecutionResult(
task_id=task.task_id,
title=task.title,
skill_used=assigned_skill_id,
artifact_path=str(artifact_file),
generated_content=clean_content,
success=True,
runtime_sec=elapsed,
)
)
logger.info("Task [%s] completed in %ss -> %s", task.task_id, elapsed, artifact_file)
return self.results
# =====================================================================
# 5. Markdown Report & Activity Log Generator
# =====================================================================
class MarkdownAuditReporter:
@staticmethod
def generate_report(
raw_prompt: str,
refined_spec: RefinedProblem,
circuit_status: CircuitBreakerStatus,
score_trajectory: List[int],
stage1_logs: List[Dict[str, Any]],
plan: Stage1PlanningOutput,
execution_results: List[TaskExecutionResult],
report_dir: Path,
) -> Path:
report_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
report_file = report_dir / f"orchestration_run_{timestamp}.md"
trajectory_str = " -> ".join(f"**{s}/10**" for s in score_trajectory)
md = f"""# Multi-Agent Workflow Execution Report
- **Run Timestamp:** {datetime.now(timezone.utc).isoformat()}
- **Circuit Breaker Status:** `{circuit_status.value.upper()}`
- **Critic Score Progression:** {trajectory_str}
- **Total Executed Tasks:** {len(execution_results)}
---
## 1. Initial Prompt & Refined Problem Specification
### Raw User Prompt
> {raw_prompt}
### Refined Objective
> {refined_spec.core_objective}
| Scope In | Scope Out | Identified Constraints | Key Assumptions |
| :--- | :--- | :--- | :--- |
| {chr(10).join(f"- {s}" for s in refined_spec.scope_in)} | {chr(10).join(f"- {s}" for s in refined_spec.scope_out)} | {chr(10).join(f"- {s}" for s in refined_spec.identified_constraints)} | {chr(10).join(f"- {s}" for s in refined_spec.key_assumptions)} |
---
## 2. Stage 1 Verification & Critic Audit Log
"""
for log in stage1_logs:
md += f"### Iteration {log.get('iteration')} Audit\n"
md += f"- **Score:** {log.get('score')}/10 (Passed: `{log.get('passed', False)}`)\n"
if log.get("findings"):
md += f"- **Findings:** {log.get('findings')}\n"
if log.get("feedback"):
md += f"- **Critique Feedback:**\n"
for fb in log["feedback"]:
md += f" - {fb}\n"
if log.get("violations"):
md += f"- **Deterministic Violations:** {', '.join(log['violations'])}\n"
md += "\n"
md += """---
## 3. Stage 1 Decomposed Execution Plan
| Task ID | Task Title | Assigned Skill | Dependencies | Target Artifact |
| :--- | :--- | :--- | :--- | :--- |
"""
for t in plan.execution_plan:
skills = ", ".join(t.recommended_skills) or "general"
deps = ", ".join(t.dependencies) or "None"
md += f"| `{t.task_id}` | {t.title} | `{skills}` | {deps} | `{t.expected_artifact_path}` |\n"
md += """
---
## 4. Stage 2 Execution Results & Generated Artifacts
"""
for res in execution_results:
md += f"### [{res.task_id}] {res.title}\n"
md += f"- **Skill Hydrated:** `{res.skill_used}`\n"
md += f"- **Output File:** `{res.artifact_path}`\n"
md += f"- **Execution Latency:** {res.runtime_sec} seconds\n\n"
md += f"```text\n{res.generated_content[:800]}...\n```\n\n"
report_file.write_text(md, encoding="utf-8")
logger.info("Audit Report successfully generated at: %s", report_file.resolve())
return report_file
# =====================================================================
# 6. Bootstrap Demo Skill Library
# =====================================================================
def bootstrap_mock_skills(skills_dir: Path):
if skills_dir.exists():
return
# Skill 1: Problem Refiner
refiner_dir = skills_dir / "planning" / "problem_refiner"
refiner_dir.mkdir(parents=True, exist_ok=True)
(refiner_dir / "SKILL.md").write_text(
"""---
id: "planning.problem_refiner"
name: "Problem Refiner"
version: "1.0.0"
stage: "planning"
description: "Isolates core goals, extracts assumptions, and defines strict system boundaries."
tags: ["refinement", "requirements"]
---
Disambiguate ambiguous phrasing, isolate negative constraints, and surface hidden dependencies.
""",
encoding="utf-8",
)
# Skill 2: Python Scaffolder
py_dir = skills_dir / "implementation" / "python_scaffolder"
py_dir.mkdir(parents=True, exist_ok=True)
(py_dir / "SKILL.md").write_text(
"""---
id: "implementation.python_scaffolder"
name: "Python Scaffolder"
version: "1.0.0"
stage: "implementation"
description: "Generates robust, modular Python modules, classes, and CLI tools."
tags: ["python", "codegen"]
---
Generate modular Python code with type annotations, docstrings, and robust error handling.
""",
encoding="utf-8",
)
# Skill 3: SQLite Builder
sql_dir = skills_dir / "implementation" / "sqlite_builder"
sql_dir.mkdir(parents=True, exist_ok=True)
(sql_dir / "SKILL.md").write_text(
"""---
id: "implementation.sqlite_builder"
name: "SQLite Schema Builder"
version: "1.0.0"
stage: "implementation"
description: "Designs relational SQL tables, indexes, and database initialization logic."
tags: ["sql", "sqlite", "database"]
---
Generate clean, idempotent SQL DDL scripts with primary keys and foreign key constraints.
""",
encoding="utf-8",
)
# =====================================================================
# 7. Main Pipeline Entrypoint
# =====================================================================
def run_orchestration(problem_statement: str, model_name: str = "llama3.1:latest"):
base_dir = Path("./orchestrator_run")
skills_dir = base_dir / "skills_library"
workspace_dir = base_dir / "outputs"
reports_dir = base_dir / "reports"
# Setup environment
bootstrap_mock_skills(skills_dir)
registry = SkillLibraryRegistry(skills_dir)
registry.scan()
ollama_client = LocalOllamaClient(model_name=model_name)
# -------------------------------------------------------------
# STAGE 1: Problem Refinement, Critic Loop, and Decomposition
# -------------------------------------------------------------
planner = Stage1Planner(client=ollama_client, registry=registry, score_threshold=7)
refined_spec, status, score_trajectory = planner.refine_with_critic_loop(problem_statement, max_iterations=3)
plan = planner.decompose(refined_spec)
# -------------------------------------------------------------
# STAGE 2: Task Execution & Artifact Generation
# -------------------------------------------------------------
executor = Stage2Executor(client=ollama_client, registry=registry, output_workspace=workspace_dir)
execution_results = executor.execute_plan(plan)
# -------------------------------------------------------------
# AUDIT LOGGING: Export Full Markdown Trace
# -------------------------------------------------------------
report_file = MarkdownAuditReporter.generate_report(
raw_prompt=problem_statement,
refined_spec=refined_spec,
circuit_status=status,
score_trajectory=score_trajectory,
stage1_logs=planner.activity_log,
plan=plan,
execution_results=execution_results,
report_dir=reports_dir,
)
print("\n" + "=" * 60)
print("PIPELINE EXECUTION COMPLETE")
print("=" * 60)
print(f"Status : {status.value}")
print(f"Refined Objective : {refined_spec.core_objective}")
print(f"Tasks Executed : {len(execution_results)}")
print(f"Artifact Workspace : {workspace_dir.resolve()}")
print(f"Markdown Audit Report: {report_file.resolve()}")
if __name__ == "__main__":
test_problem = (
"Develop an automated file ingestion agent that periodically checks an inbox folder for new JSON payload files, "
"validates them against a telemetry schema, and saves validated records into an SQLite database."
)
# Ensure Ollama is running locally: `ollama run llama3.1` or equivalent local model
run_orchestration(test_problem, model_name="llama3.1:latest")
Key Features of this Implementation
Zero External Cloud Dependencies: Powered completely via the
ollamaPython SDK pointing to your locally running weights (llama3.1,mistral,qwen2.5, etc.).Deterministic Extraction: Leverages Ollama's native JSON Schema validation (
format="json"with Pydantic JSON schemas) to ensure outputs never suffer from parser drift.Decoupled Verification Loop: Problems are isolated, evaluated by a Critic Agent, and checked against deterministic rules before any task decomposition begins.
Resilient Circuit Breaker: Detects stagnation across critic score deltas and triggers programmatic recovery patches if the score fails to advance.
Dynamic Skill Hydration: The Stage 2 Executor queries the folder tree, reads the assigned skill's
SKILL.mdinstructions, and injects them directly into the executing agent's system prompt.Self-Documenting Audit Report: Exports a timestamped Markdown report recording the initial prompt, critic scores, DAG tasks, execution latencies, and generated artifact locations.
Implement a topological sorting function in Python to ensure the planned tasks in Stage 2 strictly honor the dependencies list before execution.
"""dag_sorter.py
Deterministic topological sorting and cycle detection for PlannedTask execution graphs.
"""
from __future__ import annotations
from collections import defaultdict, deque
from typing import Dict, List, Set
from schemas import PlannedTask
class CyclicDependencyError(ValueError):
"""Raised when the execution graph contains circular task dependencies."""
pass
class MissingDependencyError(KeyError):
"""Raised when a task references a dependency that does not exist in the plan."""
pass
def sort_planned_tasks(tasks: List[PlannedTask]) -> List[PlannedTask]:
"""Sorts PlannedTasks into a strictly ordered execution sequence using Kahn's algorithm.
Validates:
1. Reference Integrity: All declared dependencies exist within the task set.
2. Graph Topology: Detects cycles and raises CyclicDependencyError with offending task IDs.
3. Determinism: Breaks zero in-degree ties using task_id to guarantee reproducible order.
"""
if not tasks:
return []
task_map: Dict[str, PlannedTask] = {t.task_id: t for t in tasks}
adjacency_list: Dict[str, List[str]] = defaultdict(list)
in_degree: Dict[str, int] = {t.task_id: 0 for t in tasks}
# Build adjacency list and compute in-degrees
for task in tasks:
for dep_id in task.dependencies:
if dep_id not in task_map:
raise MissingDependencyError(
f"Task '{task.task_id}' depends on undefined task '{dep_id}'."
)
# dep_id must complete before task.task_id
adjacency_list[dep_id].append(task.task_id)
in_degree[task.task_id] += 1
# Initialize queue with root tasks (in-degree == 0), sorted deterministically
zero_in_degree = [t_id for t_id, deg in in_degree.items() if deg == 0]
zero_in_degree.sort()
ready_queue: deque[str] = deque(zero_in_degree)
sorted_tasks: List[PlannedTask] = []
while ready_queue:
current_id = ready_queue.popleft()
sorted_tasks.append(task_map[current_id])
# Decrement in-degree for downstream dependents
next_ready = []
for dependent_id in adjacency_list[current_id]:
in_degree[dependent_id] -= 1
if in_degree[dependent_id] == 0:
next_ready.append(dependent_id)
# Sort newly unblocked nodes before appending to preserve determinism
next_ready.sort()
ready_queue.extend(next_ready)
# Cycle check: if not all nodes processed, a directed cycle exists
if len(sorted_tasks) != len(tasks):
unresolved_ids = {t_id for t_id, deg in in_degree.items() if deg > 0}
raise CyclicDependencyError(
f"Cyclic dependency detected. Unresolvable tasks: {sorted(unresolved_ids)}"
)
return sorted_tasks
Integration into Stage2Executor
Wrap the task sequence inside Stage2Executor.execute_plan() before launching agent execution:
from dag_sorter import sort_planned_tasks, CyclicDependencyError, MissingDependencyError
class Stage2Executor:
def __init__(self, client: LocalOllamaClient, registry: SkillLibraryRegistry, output_workspace: Path):
self.client = client
self.registry = registry
self.output_workspace = output_workspace
self.output_workspace.mkdir(parents=True, exist_ok=True)
self.results: List[TaskExecutionResult] = []
def execute_plan(self, plan: Stage1PlanningOutput) -> List[TaskExecutionResult]:
logger.info("Validating DAG and ordering %d planned tasks...", len(plan.execution_plan))
# 1. Enforce strict dependency ordering via Kahn's algorithm
try:
ordered_tasks = sort_planned_tasks(plan.execution_plan)
except (CyclicDependencyError, MissingDependencyError) as err:
logger.critical("DAG topological sort rejected execution plan: %s", err)
raise
logger.info(
"DAG validated. Execution order: %s",
" -> ".join(f"[{t.task_id}]" for t in ordered_tasks)
)
# 2. Execute tasks in resolved topological order
for task in ordered_tasks:
t_start = time.perf_counter()
logger.info("Executing [%s]: %s (Dependencies: %s)", task.task_id, task.title, task.dependencies)
assigned_skill_id = task.recommended_skills[0] if task.recommended_skills else "general"
skill_instructions = self.registry.get_instructions(assigned_skill_id)
agent_system_prompt = (
f"You are a Specialized Execution Agent powered by the [{assigned_skill_id}] skill.\n"
f"Your task is: {task.title}\n"
f"Description: {task.description}\n"
f"Target File: {task.expected_artifact_path}\n\n"
f"### Skill Operational Instructions:\n{skill_instructions}\n\n"
"Produce the complete, working, production-grade file content without truncated sections."
)
agent_user_prompt = (
f"Generate the exact file artifact content for '{task.expected_artifact_path}' to satisfy this task.\n"
f"System Objective: {plan.refined_problem.core_objective}\n"
f"System Constraints: {plan.refined_problem.identified_constraints}"
)
generated_content = self.client.generate_text(agent_system_prompt, agent_user_prompt)
clean_content = generated_content.strip()
if clean_content.startswith("```"):
lines = clean_content.splitlines()
if len(lines) >= 3 and lines[0].startswith("```") and lines[-1].startswith("```"):
clean_content = "\n".join(lines[1:-1])
artifact_file = self.output_workspace / task.expected_artifact_path
artifact_file.parent.mkdir(parents=True, exist_ok=True)
artifact_file.write_text(clean_content, encoding="utf-8")
elapsed = round(time.perf_counter() - t_start, 2)
self.results.append(
TaskExecutionResult(
task_id=task.task_id,
title=task.title,
skill_used=assigned_skill_id,
artifact_path=str(artifact_file),
generated_content=clean_content,
success=True,
runtime_sec=elapsed,
)
)
logger.info("Task [%s] finished in %ss -> %s", task.task_id, elapsed, artifact_file)
return self.results
Unit Verification Suite (test_dag_sorter.py)
import pytest
from schemas import PlannedTask
from dag_sorter import sort_planned_tasks, CyclicDependencyError, MissingDependencyError
def make_task(t_id: str, deps: list[str]) -> PlannedTask:
return PlannedTask(
task_id=t_id,
title=f"Title {t_id}",
description="test",
dependencies=deps,
expected_artifact_path=f"out/{t_id}.txt"
)
def test_linear_resolution():
tasks = [
make_task("TASK-03", ["TASK-02"]),
make_task("TASK-01", []),
make_task("TASK-02", ["TASK-01"]),
]
ordered = sort_planned_tasks(tasks)
assert [t.task_id for t in ordered] == ["TASK-01", "TASK-02", "TASK-03"]
def test_branching_dag():
tasks = [
make_task("TASK-04", ["TASK-02", "TASK-03"]),
make_task("TASK-03", ["TASK-01"]),
make_task("TASK-02", ["TASK-01"]),
make_task("TASK-01", []),
]
ordered = sort_planned_tasks(tasks)
assert ordered[0].task_id == "TASK-01"
assert ordered[-1].task_id == "TASK-04"
assert {ordered[1].task_id, ordered[2].task_id} == {"TASK-02", "TASK-03"}
def test_cycle_detection():
tasks = [
make_task("TASK-01", ["TASK-02"]),
make_task("TASK-02", ["TASK-01"]),
]
with pytest.raises(CyclicDependencyError) as exc:
sort_planned_tasks(tasks)
assert "TASK-01" in str(exc.value) and "TASK-02" in str(exc.value)
def test_missing_dependency():
tasks = [
make_task("TASK-01", ["NON_EXISTENT_TASK"]),
]
with pytest.raises(MissingDependencyError):
sort_planned_tasks(tasks)Update the Stage 2 agent prompt hydration to read and inject the generated file content of predecessor tasks.
Predecessor Content Injection Architecture
When executing tasks in topological sequence, a downstream agent often cannot produce working code or schemas without inspecting the exact interfaces, class names, and tables established by its upstream dependencies.
To prevent context bloat and token exhaustion on local models, the updated hydration mechanism:
Resolves all immediate parent dependencies declared in
task.dependencies.Verifies the generated predecessor artifact exists on disk in the workspace.
Injects the predecessor artifact's relative path and content into the downstream agent's prompt envelope under a clear
### UPSTREAM ARTIFACT CONTEXTblock.Implements safe truncation thresholds (with head/tail preservation) if an upstream artifact exceeds local context limits.
[TASK-01: Schemas] ──────► Writes `schemas.py` to workspace
│
▼ (Direct Dependency)
[TASK-02: DB Builder] ──► Reads `schemas.py` from disk
Hydrates into TASK-02 prompt ──► Writes `db.py`
Updated Stage2Executor Implementation
Replace the Stage2Executor class in your application with the following updated implementation:
"""stage2_executor.py
Stage 2 Execution Orchestrator with upstream artifact inspection and prompt hydration.
"""
from __future__ import annotations
import logging
import time
from pathlib import Path
from typing import Dict, List, Optional
from schemas import PlannedTask, Stage1PlanningOutput, TaskExecutionResult
from dag_sorter import sort_planned_tasks
logger = logging.getLogger("Stage2Executor")
class Stage2Executor:
"""Executes topologically ordered tasks and injects upstream artifact contents."""
def __init__(
self,
client: LocalOllamaClient,
registry: SkillLibraryRegistry,
output_workspace: Path,
max_predecessor_chars: int = 4000,
):
self.client = client
self.registry = registry
self.output_workspace = output_workspace
self.output_workspace.mkdir(parents=True, exist_ok=True)
self.max_predecessor_chars = max_predecessor_chars
self.results: List[TaskExecutionResult] = []
# In-memory index of completed tasks: task_id -> artifact Path
self.completed_artifacts: Dict[str, Path] = {}
def _truncate_content(self, text: str, max_chars: int) -> str:
"""Preserves top and bottom context if an upstream file exceeds char limits."""
if len(text) <= max_chars:
return text
slice_size = max_chars // 2
return (
text[:slice_size]
+ f"\n\n... [TRUNCATED {len(text) - max_chars} CHARACTERS] ...\n\n"
+ text[-slice_size:]
)
def _build_predecessor_context(self, task: PlannedTask) -> str:
"""Reads and formats generated content from all direct dependency tasks."""
if not task.dependencies:
return ""
context_blocks: List[str] = []
for dep_id in task.dependencies:
artifact_file = self.completed_artifacts.get(dep_id)
if not artifact_file or not artifact_file.exists():
logger.warning(
"Task [%s] declared dependency [%s], but artifact on disk was not found.",
task.task_id,
dep_id,
)
continue
try:
raw_content = artifact_file.read_text(encoding="utf-8")
truncated = self._truncate_content(raw_content, self.max_predecessor_chars)
rel_path = artifact_file.relative_to(self.output_workspace)
block = (
f"#### Upstream Artifact from Dependency [{dep_id}] (`{rel_path}`):\n"
f"```text\n{truncated}\n```"
)
context_blocks.append(block)
except Exception as err:
logger.error("Failed reading upstream artifact %s: %s", artifact_file, err)
if not context_blocks:
return ""
return (
"\n### UPSTREAM ARTIFACT CONTEXT (Generated Predecessor Dependencies):\n"
"You MUST align your implementation strictly with the classes, functions, schemas, "
"and configurations established in the following upstream artifacts:\n\n"
+ "\n\n".join(context_blocks)
+ "\n"
)
def execute_plan(self, plan: Stage1PlanningOutput) -> List[TaskExecutionResult]:
logger.info("Sorting %d planned tasks via topological sort...", len(plan.execution_plan))
ordered_tasks = sort_planned_tasks(plan.execution_plan)
logger.info(
"Topological execution sequence: %s",
" -> ".join(f"[{t.task_id}]" for t in ordered_tasks),
)
for task in ordered_tasks:
t_start = time.perf_counter()
logger.info("Executing [%s]: %s", task.task_id, task.title)
# 1. Resolve Skill Instructions
assigned_skill_id = task.recommended_skills[0] if task.recommended_skills else "general"
skill_instructions = self.registry.get_instructions(assigned_skill_id)
# 2. Extract and Hydrate Predecessor Artifacts
predecessor_context = self._build_predecessor_context(task)
# 3. Formulate Agent Prompts
agent_system_prompt = (
f"You are a Specialized Execution Agent powered by the [{assigned_skill_id}] skill.\n"
f"Your task: {task.title}\n"
f"Description: {task.description}\n"
f"Target File: {task.expected_artifact_path}\n\n"
f"### Skill Operational Instructions:\n{skill_instructions}\n\n"
"Produce the complete, working, production-grade file content without truncated sections."
)
agent_user_prompt = (
f"Generate the exact file artifact content for '{task.expected_artifact_path}' to satisfy this task.\n\n"
f"### Overall Problem Specification:\n"
f"- Objective: {plan.refined_problem.core_objective}\n"
f"- Scope In: {', '.join(plan.refined_problem.scope_in)}\n"
f"- Constraints: {', '.join(plan.refined_problem.identified_constraints)}\n"
f"{predecessor_context}\n"
"Respond ONLY with the complete file contents to be written to disk."
)
# 4. Generate Content via Local LLM
generated_content = self.client.generate_text(agent_system_prompt, agent_user_prompt)
# Clean markdown code fences if wrapped
clean_content = generated_content.strip()
if clean_content.startswith("```"):
lines = clean_content.splitlines()
if len(lines) >= 3 and lines[0].startswith("```") and lines[-1].startswith("```"):
clean_content = "\n".join(lines[1:-1])
# 5. Write Concrete File to Workspace
artifact_file = self.output_workspace / task.expected_artifact_path
artifact_file.parent.mkdir(parents=True, exist_ok=True)
artifact_file.write_text(clean_content, encoding="utf-8")
# Index artifact for subsequent dependents
self.completed_artifacts[task.task_id] = artifact_file
elapsed = round(time.perf_counter() - t_start, 2)
self.results.append(
TaskExecutionResult(
task_id=task.task_id,
title=task.title,
skill_used=assigned_skill_id,
artifact_path=str(artifact_file),
generated_content=clean_content,
success=True,
runtime_sec=elapsed,
)
)
logger.info("Task [%s] finished in %ss -> %s", task.task_id, elapsed, artifact_file)
return self.results
Key Operational Improvements
Interface Consistency: Downstream implementation agents (like an SQLite repository or API router) directly observe the data classes or table definitions written by preceding tasks, preventing name and type mismatches.
Deterministic Disk-Backed State: Content is retrieved directly from the verified file system artifact rather than relying on LLM chat memory or lossy intermediate summaries.
Context Budget Protection:
max_predecessor_charsbounds context overhead, ensuring prompt lengths stay within the context window limits of local Ollama models.
Add a self-correction loop in Stage 2 that feeds execution errors or syntax exceptions back to the agent with the prior file content to generate a fix.
Self-Correction Architecture
The Stage 2 self-correction loop catches syntax or execution faults immediately upon artifact generation. Rather than propagating broken code downstream to dependent tasks, it subjects the newly created file to automated language-specific verification (e.g., Python ast.parse / py_compile, JSON schema validation, or SQLite DDL dry-runs). If an exception occurs, the error traceback, exact line number, and broken code are fed back into the agent for targeted repair iterations.
[Topological Task: TASK-N]
│
▼
[Agent Generates Artifact]
│
▼
[Deterministic Validator]
├── Python: ast.parse() / py_compile
├── JSON: json.loads()
└── SQL: sqlite3.connect(":memory:").executescript()
│
┌──────────┴──────────┐
(Success) (Exception)
│ │
▼ ▼
[Index Artifact & [Self-Correction Loop]
Proceed to Next] ├── Feed Traceback + Line Error + Faulty Code
├── Prompt: "Fix the syntax/runtime fault"
└── Retry (Up to max_retries)
Implementation: Validator and Self-Correcting Executor
Save or update this module as stage2_executor.py:
"""stage2_executor.py
Stage 2 Execution Orchestrator featuring:
- Topologically ordered task sequencing
- Upstream artifact context injection
- In-process deterministic syntax/runtime validation
- Automated self-correction feedback loop on failure
"""
from __future__ import annotations
import ast
import json
import logging
import sqlite3
import time
import traceback
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from dag_sorter import sort_planned_tasks
from schemas import PlannedTask, Stage1PlanningOutput, TaskExecutionResult
logger = logging.getLogger("Stage2Executor")
class ArtifactSyntaxValidator:
"""Zero-dependency deterministic syntax and structure validation."""
@staticmethod
def validate(file_path: Path, content: str) -> Tuple[bool, Optional[str]]:
suffix = file_path.suffix.lower()
# 1. Python Syntax Validation
if suffix == ".py":
try:
ast.parse(content, filename=str(file_path))
return True, None
except SyntaxError as e:
return False, f"Python SyntaxError at line {e.lineno}, col {e.offset}: {e.msg}\nCode segment: {e.text}"
except Exception as e:
return False, f"Python AST Parse Error: {str(e)}\n{traceback.format_exc()}"
# 2. JSON Validation
elif suffix == ".json":
try:
json.loads(content)
return True, None
except json.JSONDecodeError as e:
return False, f"JSONDecodeError at line {e.lineno}, col {e.colno}: {e.msg}"
# 3. SQLite DDL / SQL Script In-Memory Dry Run
elif suffix == ".sql":
con = sqlite3.connect(":memory:")
try:
con.executescript(content)
return True, None
except sqlite3.Error as e:
return False, f"SQLite Execution Error: {str(e)}"
finally:
con.close()
# General text / markdown files pass by default
return True, None
class Stage2Executor:
"""Executes planned tasks with context hydration and self-correcting validation."""
def __init__(
self,
client, # LocalOllamaClient instance
registry,
output_workspace: Path,
max_predecessor_chars: int = 4000,
max_fix_retries: int = 3,
):
self.client = client
self.registry = registry
self.output_workspace = output_workspace
self.output_workspace.mkdir(parents=True, exist_ok=True)
self.max_predecessor_chars = max_predecessor_chars
self.max_fix_retries = max_fix_retries
self.results: List[TaskExecutionResult] = []
self.completed_artifacts: Dict[str, Path] = {}
def _clean_markdown_fences(self, content: str) -> str:
clean = content.strip()
if clean.startswith("```"):
lines = clean.splitlines()
if len(lines) >= 3 and lines[0].startswith("```") and lines[-1].startswith("```"):
clean = "\n".join(lines[1:-1])
return clean.strip()
def _truncate_content(self, text: str, max_chars: int) -> str:
if len(text) <= max_chars:
return text
half = max_chars // 2
return text[:half] + f"\n\n... [TRUNCATED {len(text) - max_chars} CHARACTERS] ...\n\n" + text[-half:]
def _build_predecessor_context(self, task: PlannedTask) -> str:
if not task.dependencies:
return ""
context_blocks = []
for dep_id in task.dependencies:
artifact_file = self.completed_artifacts.get(dep_id)
if not artifact_file or not artifact_file.exists():
logger.warning("Task [%s] missing artifact for dep [%s]", task.task_id, dep_id)
continue
try:
raw_text = artifact_file.read_text(encoding="utf-8")
truncated = self._truncate_content(raw_text, self.max_predecessor_chars)
rel_path = artifact_file.relative_to(self.output_workspace)
context_blocks.append(
f"#### Upstream Artifact from [{dep_id}] (`{rel_path}`):\n```text\n{truncated}\n```"
)
except Exception as err:
logger.error("Failed reading upstream artifact %s: %s", artifact_file, err)
if not context_blocks:
return ""
return (
"\n### UPSTREAM ARTIFACT CONTEXT (Predecessor Dependencies):\n"
"Align your code strictly with types, classes, functions, and schemas defined below:\n\n"
+ "\n\n".join(context_blocks)
+ "\n"
)
def _generate_remedy(
self,
task: PlannedTask,
assigned_skill_id: str,
skill_instructions: str,
flawed_content: str,
error_message: str,
attempt: int,
) -> str:
"""Prompts the local model to correct the broken code with error tracebacks."""
logger.info("Attempting self-correction iteration %d/%d for [%s]...", attempt, self.max_fix_retries, task.task_id)
system_prompt = (
f"You are an Expert Automated Code Repair Agent using skill [{assigned_skill_id}].\n"
f"Target File: {task.expected_artifact_path}\n"
f"{skill_instructions}\n\n"
"Your previous code generation attempt produced a syntax or execution error.\n"
"Inspect the error traceback and prior broken code, resolve the root cause, and return the complete corrected file.\n"
"Return ONLY the corrected, raw code to be written directly to disk. Do NOT truncate."
)
user_prompt = (
f"### ERROR TRACEBACK ENCOUNTERED:\n{error_message}\n\n"
f"### PRIOR FAULTY FILE CONTENT:\n```text\n{flawed_content}\n```\n\n"
"Provide the complete, functional, fixed file content resolving the error above."
)
remedy_raw = self.client.generate_text(system_prompt, user_prompt)
return self._clean_markdown_fences(remedy_raw)
def execute_plan(self, plan: Stage1PlanningOutput) -> List[TaskExecutionResult]:
logger.info("Sorting %d planned tasks via topological sort...", len(plan.execution_plan))
ordered_tasks = sort_planned_tasks(plan.execution_plan)
for task in ordered_tasks:
t_start = time.perf_counter()
logger.info("Executing [%s]: %s", task.task_id, task.title)
assigned_skill_id = task.recommended_skills[0] if task.recommended_skills else "general"
skill_instructions = self.registry.get_instructions(assigned_skill_id)
predecessor_context = self._build_predecessor_context(task)
# Initial Generation Prompt
agent_system_prompt = (
f"You are a Specialized Execution Agent powered by the [{assigned_skill_id}] skill.\n"
f"Task: {task.title}\n"
f"Description: {task.description}\n"
f"Target File: {task.expected_artifact_path}\n\n"
f"### Skill Operational Instructions:\n{skill_instructions}\n\n"
"Produce the complete, functional, production-ready file content without truncation."
)
agent_user_prompt = (
f"Generate the exact file artifact content for '{task.expected_artifact_path}'.\n\n"
f"### System Specification:\n"
f"- Objective: {plan.refined_problem.core_objective}\n"
f"- Constraints: {', '.join(plan.refined_problem.identified_constraints)}\n"
f"{predecessor_context}\n"
"Respond ONLY with the complete file contents to be written to disk."
)
generated_content = self.client.generate_text(agent_system_prompt, agent_user_prompt)
clean_content = self._clean_markdown_fences(generated_content)
artifact_file = self.output_workspace / task.expected_artifact_path
artifact_file.parent.mkdir(parents=True, exist_ok=True)
# Validation & Self-Correction Loop
is_valid, error_msg = ArtifactSyntaxValidator.validate(artifact_file, clean_content)
retry_count = 0
while not is_valid and retry_count < self.max_fix_retries:
retry_count += 1
logger.warning(
"Task [%s] failed validation on attempt %d: %s",
task.task_id,
retry_count,
error_msg.splitlines()[0] if error_msg else "Unknown",
)
clean_content = self._generate_remedy(
task=task,
assigned_skill_id=assigned_skill_id,
skill_instructions=skill_instructions,
flawed_content=clean_content,
error_message=error_msg or "Unknown validation error",
attempt=retry_count,
)
is_valid, error_msg = ArtifactSyntaxValidator.validate(artifact_file, clean_content)
if not is_valid:
logger.critical(
"Task [%s] failed all %d self-correction attempts: %s",
task.task_id,
self.max_fix_retries,
error_msg,
)
raise RuntimeError(
f"Execution halt: Task [{task.task_id}] failed self-correction.\nError: {error_msg}"
)
# Persist validated artifact to disk
artifact_file.write_text(clean_content, encoding="utf-8")
self.completed_artifacts[task.task_id] = artifact_file
elapsed = round(time.perf_counter() - t_start, 2)
self.results.append(
TaskExecutionResult(
task_id=task.task_id,
title=task.title,
skill_used=assigned_skill_id,
artifact_path=str(artifact_file),
generated_content=clean_content,
success=True,
runtime_sec=elapsed,
)
)
logger.info(
"Task [%s] verified & written (Retries: %d, Elapsed: %ss) -> %s",
task.task_id,
retry_count,
elapsed,
artifact_file,
)
return self.results
Verification and Edge Case Handling
Python
ast.parseSafety: Catches missing colons, unclosed brackets, indentation errors, and malformed strings before saving the file or executing downstream tasks.SQLite In-Memory Dry Run: Catches missing table commas, bad foreign key declarations, invalid datatypes, and reserved keyword collisions using standard
sqlite3.connect(":memory:").Bounded Iteration Count: Bound by
max_fix_retries(default: 3) to prevent runaway inference loops if a model gets stuck on a syntax edge case.
Update the Markdown audit logger to include a dedicated section detailing all self-correction attempts, error tracebacks, and applied diffs.
Self-Correction Audit & Diff Architecture
To capture self-correction iterations transparently, the audit subsystem tracks:
Pre-fix code snapshot before every remediation attempt.
Deterministic error traceback (line numbers, syntax tokens, or database exceptions).
Unified Git-style diff computed via Python's standard
difflib.unified_diff.Structured telemetry per correction event injected directly into the final Markdown report.
1. Updated Schemas (schemas.py)
Add the correction event schema and wire it to TaskExecutionResult:
from __future__ import annotations
from typing import List, Optional
from pydantic import BaseModel, Field
class CorrectionEvent(BaseModel):
attempt: int
error_message: str
pre_fix_snippet: str
post_fix_snippet: str
unified_diff: str
timestamp_utc: str
class TaskExecutionResult(BaseModel):
task_id: str
title: str
skill_used: str
artifact_path: str
generated_content: str
success: bool
runtime_sec: float
correction_history: List[CorrectionEvent] = Field(default_factory=list)
2. Instrumented Stage2Executor with Diff Tracking (stage2_executor.py)
This updated execution loop tracks the faulty code, captures the exception, prompts the local Ollama repair agent, computes a unified diff, and attaches the audit events to TaskExecutionResult:
"""stage2_executor.py
Stage 2 Execution Orchestrator with predecessor hydration, deterministic validation,
automated repair loops, and unified git-style diff tracking.
"""
from __future__ import annotations
import difflib
import logging
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional
from dag_sorter import sort_planned_tasks
from schemas import (
CorrectionEvent,
PlannedTask,
Stage1PlanningOutput,
TaskExecutionResult,
)
logger = logging.getLogger("Stage2Executor")
class Stage2Executor:
def __init__(
self,
client, # LocalOllamaClient instance
registry,
output_workspace: Path,
max_predecessor_chars: int = 4000,
max_fix_retries: int = 3,
):
self.client = client
self.registry = registry
self.output_workspace = output_workspace
self.output_workspace.mkdir(parents=True, exist_ok=True)
self.max_predecessor_chars = max_predecessor_chars
self.max_fix_retries = max_fix_retries
self.results: List[TaskExecutionResult] = []
self.completed_artifacts: Dict[str, Path] = {}
def _clean_markdown_fences(self, content: str) -> str:
clean = content.strip()
if clean.startswith("```"):
lines = clean.splitlines()
if len(lines) >= 3 and lines[0].startswith("```") and lines[-1].startswith("```"):
clean = "\n".join(lines[1:-1])
return clean.strip()
def _compute_diff(self, old_text: str, new_text: str, filename: str) -> str:
"""Computes a standard unified diff between faulty and repaired code."""
old_lines = old_text.splitlines(keepends=True)
new_lines = new_text.splitlines(keepends=True)
diff = difflib.unified_diff(
old_lines,
new_lines,
fromfile=f"a/{filename} (faulty)",
tofile=f"b/{filename} (repaired)",
lineterm="",
)
return "".join(diff).strip()
def _generate_remedy(
self,
task: PlannedTask,
assigned_skill_id: str,
skill_instructions: str,
flawed_content: str,
error_message: str,
attempt: int,
) -> str:
logger.info("Executing self-correction attempt %d/%d for [%s]...", attempt, self.max_fix_retries, task.task_id)
system_prompt = (
f"You are an Expert Automated Code Repair Agent using skill [{assigned_skill_id}].\n"
f"Target File: {task.expected_artifact_path}\n"
f"{skill_instructions}\n\n"
"Your previous code generation attempt failed validation with a syntax or execution error.\n"
"Inspect the error traceback and prior broken code, resolve the root cause, and return the complete corrected file.\n"
"Return ONLY the raw corrected file contents. Do not truncate sections or omit existing working logic."
)
user_prompt = (
f"### ERROR ENCOUNTERED:\n{error_message}\n\n"
f"### PRIOR FAULTY FILE CONTENT:\n```text\n{flawed_content}\n```\n\n"
"Provide the complete, functional, fixed file content resolving the error above."
)
remedy_raw = self.client.generate_text(system_prompt, user_prompt)
return self._clean_markdown_fences(remedy_raw)
def execute_plan(self, plan: Stage1PlanningOutput) -> List[TaskExecutionResult]:
from stage2_validator import ArtifactSyntaxValidator # External validator
logger.info("Topologically ordering and executing %d planned tasks...", len(plan.execution_plan))
ordered_tasks = sort_planned_tasks(plan.execution_plan)
for task in ordered_tasks:
t_start = time.perf_counter()
logger.info("Executing [%s]: %s", task.task_id, task.title)
assigned_skill_id = task.recommended_skills[0] if task.recommended_skills else "general"
skill_instructions = self.registry.get_instructions(assigned_skill_id)
predecessor_context = self._build_predecessor_context(task)
agent_system_prompt = (
f"You are a Specialized Execution Agent powered by the [{assigned_skill_id}] skill.\n"
f"Task: {task.title}\n"
f"Description: {task.description}\n"
f"Target File: {task.expected_artifact_path}\n\n"
f"### Skill Operational Instructions:\n{skill_instructions}\n\n"
"Produce the complete, functional, production-ready file content without truncation."
)
agent_user_prompt = (
f"Generate the exact file artifact content for '{task.expected_artifact_path}'.\n\n"
f"### System Specification:\n"
f"- Objective: {plan.refined_problem.core_objective}\n"
f"- Constraints: {', '.join(plan.refined_problem.identified_constraints)}\n"
f"{predecessor_context}\n"
"Respond ONLY with the complete file contents to be written to disk."
)
generated_content = self.client.generate_text(agent_system_prompt, agent_user_prompt)
current_content = self._clean_markdown_fences(generated_content)
artifact_file = self.output_workspace / task.expected_artifact_path
artifact_file.parent.mkdir(parents=True, exist_ok=True)
# Validation & Self-Correction Loop with Diff Logging
is_valid, error_msg = ArtifactSyntaxValidator.validate(artifact_file, current_content)
correction_history: List[CorrectionEvent] = []
retry_count = 0
while not is_valid and retry_count < self.max_fix_retries:
retry_count += 1
prior_flawed_content = current_content
logger.warning(
"Task [%s] failed validation on attempt %d: %s",
task.task_id,
retry_count,
error_msg.splitlines()[0] if error_msg else "Unknown",
)
# Generate fixed content
current_content = self._generate_remedy(
task=task,
assigned_skill_id=assigned_skill_id,
skill_instructions=skill_instructions,
flawed_content=prior_flawed_content,
error_message=error_msg or "Unknown validation error",
attempt=retry_count,
)
# Compute unified diff
diff_text = self._compute_diff(
old_text=prior_flawed_content,
new_text=current_content,
filename=task.expected_artifact_path,
)
correction_history.append(
CorrectionEvent(
attempt=retry_count,
error_message=error_msg or "Validation Failure",
pre_fix_snippet=prior_flawed_content[:600],
post_fix_snippet=current_content[:600],
unified_diff=diff_text,
timestamp_utc=datetime.now(timezone.utc).isoformat(),
)
)
is_valid, error_msg = ArtifactSyntaxValidator.validate(artifact_file, current_content)
if not is_valid:
logger.critical("Task [%s] failed all %d repair attempts.", task.task_id, self.max_fix_retries)
raise RuntimeError(f"Halted: Task [{task.task_id}] failed self-correction.\nError: {error_msg}")
# Persist validated artifact to disk
artifact_file.write_text(current_content, encoding="utf-8")
self.completed_artifacts[task.task_id] = artifact_file
elapsed = round(time.perf_counter() - t_start, 2)
self.results.append(
TaskExecutionResult(
task_id=task.task_id,
title=task.title,
skill_used=assigned_skill_id,
artifact_path=str(artifact_file),
generated_content=current_content,
success=True,
runtime_sec=elapsed,
correction_history=correction_history,
)
)
return self.results
def _build_predecessor_context(self, task: PlannedTask) -> str:
if not task.dependencies:
return ""
blocks = []
for dep_id in task.dependencies:
artifact_file = self.completed_artifacts.get(dep_id)
if artifact_file and artifact_file.exists():
raw_text = artifact_file.read_text(encoding="utf-8")
rel_path = artifact_file.relative_to(self.output_workspace)
blocks.append(f"#### Upstream Artifact from [{dep_id}] (`{rel_path}`):\n```text\n{raw_text[:self.max_predecessor_chars]}\n```")
return "\n### UPSTREAM ARTIFACT CONTEXT:\n" + "\n\n".join(blocks) if blocks else ""
3. Markdown Audit Reporter (markdown_reporter.py)
This updated reporter formats Stage 1, Stage 2, and renders a dedicated Section 5 for Self-Correction & Automated Repair Logs featuring the exact error tracebacks and git-style unified diff blocks:
"""markdown_reporter.py
Renders comprehensive, audit-ready Markdown run logs including self-correction diffs.
"""
from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List
from schemas import (
CircuitBreakerStatus,
PlannedTask,
RefinedProblem,
Stage1PlanningOutput,
TaskExecutionResult,
)
class MarkdownAuditReporter:
@staticmethod
def generate_report(
raw_prompt: str,
refined_spec: RefinedProblem,
circuit_status: CircuitBreakerStatus,
score_trajectory: List[int],
stage1_logs: List[Dict[str, Any]],
plan: Stage1PlanningOutput,
execution_results: List[TaskExecutionResult],
report_dir: Path,
) -> Path:
report_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
report_file = report_dir / f"orchestration_run_{timestamp}.md"
total_corrections = sum(len(res.correction_history) for res in execution_results)
trajectory_str = " -> ".join(f"**{s}/10**" for s in score_trajectory)
md = f"""# Multi-Agent Workflow Execution & Audit Report
- **Run Timestamp:** `{datetime.now(timezone.utc).isoformat()}`
- **Circuit Breaker Status:** `{circuit_status.value.upper()}`
- **Critic Score Progression:** {trajectory_str}
- **Stage 2 Self-Correction Events:** `{total_corrections}`
- **Artifacts Delivered:** `{len(execution_results)}`
---
## 1. Problem Specification & Scope Boundaries
### Raw User Prompt
> {raw_prompt}
### Refined Core Objective
> {refined_spec.core_objective}
| Scope In | Scope Out | Identified Constraints | Key Assumptions |
| :--- | :--- | :--- | :--- |
| {chr(10).join(f"- {s}" for s in refined_spec.scope_in)} | {chr(10).join(f"- {s}" for s in refined_spec.scope_out)} | {chr(10).join(f"- {s}" for s in refined_spec.identified_constraints)} | {chr(10).join(f"- {s}" for s in refined_spec.key_assumptions)} |
---
## 2. Stage 1 Verification & Critic Evaluation Trail
"""
for log in stage1_logs:
md += f"### Iteration {log.get('iteration')} Audit\n"
md += f"- **Score:** `{log.get('score')}/10` (Passed: `{log.get('passed', False)}`)\n"
if log.get("findings"):
md += f"- **Findings:** {log.get('findings')}\n"
if log.get("feedback"):
md += f"- **Actionable Feedback:**\n"
for fb in log["feedback"]:
md += f" - {fb}\n"
if log.get("violations"):
md += f"- **Deterministic Violations:** {', '.join(log['violations'])}\n"
md += "\n"
md += """---
## 3. Stage 1 Execution Plan (DAG)
| Task ID | Task Title | Assigned Skill | Dependencies | Target File |
| :--- | :--- | :--- | :--- | :--- |
"""
for t in plan.execution_plan:
skills = ", ".join(t.recommended_skills) or "general"
deps = ", ".join(t.dependencies) or "None"
md += f"| `{t.task_id}` | {t.title} | `{skills}` | {deps} | `{t.expected_artifact_path}` |\n"
md += """
---
## 4. Stage 2 Execution Results
| Task ID | Status | Hydrated Skill | Latency | Self-Corrections | Target Artifact |
| :--- | :--- | :--- | :--- | :--- | :--- |
"""
for res in execution_results:
status_badge = "✅ Passed" if res.success else "❌ Failed"
corrections_count = len(res.correction_history)
md += f"| `{res.task_id}` | {status_badge} | `{res.skill_used}` | {res.runtime_sec}s | {corrections_count} | `{res.artifact_path}` |\n"
md += """
---
## 5. Self-Correction & Automated Repair Trail
"""
if total_corrections == 0:
md += "_No syntax or execution errors occurred. All artifacts passed initial verification on first attempt._\n\n"
else:
for res in execution_results:
if not res.correction_history:
continue
md += f"### Task [{res.task_id}]: {res.title}\n"
md += f"- **Artifact Path:** `{res.artifact_path}`\n"
md += f"- **Total Repair Passes:** `{len(res.correction_history)}`\n\n"
for event in res.correction_history:
md += f"#### Attempt #{event.attempt} Remediation Trace (`{event.timestamp_utc}`)\n\n"
md += f"**Triggering Exception / Syntax Traceback:**\n"
md += f"```text\n{event.error_message.strip()}\n```\n\n"
md += f"**Applied Code Diff:**\n"
md += f"```diff\n{event.unified_diff if event.unified_diff else '# No lines changed'}\n```\n\n"
md += """---
## 6. Generated Artifact Code Listings
"""
for res in execution_results:
md += f"### `{res.artifact_path}`\n"
md += f"```python\n{res.generated_content.strip()}\n```\n\n"
report_file.write_text(md, encoding="utf-8")
return report_file
Sample Rendered Output in Section 5
When a task encounters an error (such as a missing colon or an unmatched parenthesis in Python), Section 5 of the generated Markdown displays:
### Task [TASK-02]: Database Schema Migrations
- **Artifact Path:** `/workspace/outputs/db/schema.sql`
- **Total Repair Passes:** `1`
#### Attempt #1 Remediation Trace (`2026-11-04T18:24:10.512Z`)
**Triggering Exception / Syntax Traceback:**
```text
SQLite Execution Error: near "AUTOINCREMENT": syntax error at line 4
Applied Code Diff:
--- a/db/schema.sql (faulty)
+++ b/db/schema.sql (repaired)
@@ -1,6 +1,6 @@
CREATE TABLE IF NOT EXISTS telemetry_records (
- id INT AUTOINCREMENT,
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
device_id TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);
Next options for auditing and system resilience:
Add colorized terminal diff formatting with Rich
query="Implement a CLI visualization using the Rich library to display colorized git diffs and task execution tables in the terminal as they happen."/>
Add static type checking with Mypy in Stage 2
query="Extend the Stage 2 validation step to run mypy static type checking against generated Python files and feed type errors into the self-correction loop."/>
Want to add Mypy or Flake8 validation to catch subtle type and style errors before diffing?
query="Implement a Mypy/Flake8 validator in Stage 2 so that type-check errors are also intercepted, diffed, and corrected before final delivery."/>Bring it altogether as one Python application that can be run end-to-end
Here is the complete, single-file, end-to-end Python application. It bundles:
Self-Bootstrapping Directory Skill Library: Automatically lays out
planning/andimplementation/folders with self-containedSKILL.mdmanifests.Stage 1 (Planning Orchestrator): Executes the restate and refine pass, deterministic boundary validation, the Evaluator-Critic loop, and the stagnation/exhaustion circuit breaker.
Deterministic Graph Resolution: Decomposes the refined specification and enforces Kahn’s topological sort on task dependencies.
Stage 2 (Execution Orchestrator): Injects upstream artifact context, queries local Ollama models, validates code syntax (
ast.parse,sqlite3,json), and executes the automated self-correction loop.Unified Diff & Markdown Reporter: Calculates git-style diffs with
diffliband exports the full trace to a timestamped Markdown audit report.
"""app.py
End-to-End Autonomous Multi-Agent Orchestrator.
Runs locally using Ollama, executing Stage 1 planning, DAG sorting,
Stage 2 execution with self-correction, and Markdown audit export.
"""
from __future__ import annotations
import ast
from collections import defaultdict, deque
from datetime import datetime, timezone
import difflib
from enum import Enum
import json
import logging
from pathlib import Path
import sqlite3
import sys
import time
import traceback
from typing import Any, Dict, List, Optional, Tuple
import ollama
from pydantic import BaseModel, Field
import yaml
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("OrchestratorApp")
# =====================================================================
# 1. Unified Pydantic Contracts
# =====================================================================
class OrchestrationStage(str, Enum):
PLANNING = "planning"
IMPLEMENTATION = "implementation"
VERIFICATION = "verification"
GENERAL = "general"
class CircuitBreakerStatus(str, Enum):
CLOSED = "closed"
RECOVERED = "recovered"
QUARANTINED = "quarantined"
class SkillMetadata(BaseModel):
id: str
name: str
version: str = "1.0.0"
stage: OrchestrationStage = OrchestrationStage.GENERAL
description: str
tags: List[str] = Field(default_factory=list)
class RefinedProblem(BaseModel):
core_objective: str = Field(..., description="Precise statement of goal")
scope_in: List[str] = Field(default_factory=list)
scope_out: List[str] = Field(default_factory=list)
identified_constraints: List[str] = Field(default_factory=list)
key_assumptions: List[str] = Field(default_factory=list)
class CriticEvaluationResult(BaseModel):
overall_score: int = Field(..., ge=1, le=10)
passed: bool
findings: str
actionable_feedback: List[str] = Field(default_factory=list)
class PlannedTask(BaseModel):
task_id: str = Field(..., description="E.g., TASK-01")
title: str
description: str
target_stage: OrchestrationStage = OrchestrationStage.IMPLEMENTATION
recommended_skills: List[str] = Field(default_factory=list)
dependencies: List[str] = Field(default_factory=list)
expected_artifact_path: str = Field(..., description="Relative file path to generate")
class Stage1PlanningOutput(BaseModel):
refined_problem: RefinedProblem
execution_plan: List[PlannedTask]
audit_notes: str
class CorrectionEvent(BaseModel):
attempt: int
error_message: str
pre_fix_snippet: str
post_fix_snippet: str
unified_diff: str
timestamp_utc: str
class TaskExecutionResult(BaseModel):
task_id: str
title: str
skill_used: str
artifact_path: str
generated_content: str
success: bool
runtime_sec: float
correction_history: List[CorrectionEvent] = Field(default_factory=list)
# =====================================================================
# 2. Local Ollama Client Wrapper
# =====================================================================
class LocalOllamaClient:
def __init__(self, model_name: str = "llama3.1:latest"):
self.model_name = model_name
def generate_structured(self, system_prompt: str, user_prompt: str, schema_class: type[BaseModel]) -> BaseModel:
schema_json = json.dumps(schema_class.model_json_schema())
augmented_system = (
f"{system_prompt}\n\n"
f"You MUST reply with valid JSON matching this JSON Schema exactly:\n{schema_json}\n"
"Do not emit explanatory text, introductory remarks, or markdown code fences outside JSON."
)
response = ollama.chat(
model=self.model_name,
messages=[
{"role": "system", "content": augmented_system},
{"role": "user", "content": user_prompt},
],
format="json",
options={"temperature": 0.1},
)
content = response["message"]["content"]
return schema_class.model_validate_json(content)
def generate_text(self, system_prompt: str, user_prompt: str) -> str:
response = ollama.chat(
model=self.model_name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
options={"temperature": 0.2},
)
return response["message"]["content"]
# =====================================================================
# 3. Skill Library Registry
# =====================================================================
class SkillLibraryRegistry:
def __init__(self, root_dir: Path):
self.root_dir = root_dir
self.skills: Dict[str, SkillMetadata] = {}
self.instructions: Dict[str, str] = {}
def scan(self) -> None:
self.skills.clear()
self.instructions.clear()
if not self.root_dir.exists():
return
for skill_path in self.root_dir.glob("**/SKILL.md"):
try:
raw_text = skill_path.read_text(encoding="utf-8")
parts = raw_text.split("---", 2)
if len(parts) < 3:
continue
meta_dict = yaml.safe_load(parts[1])
meta = SkillMetadata(**meta_dict)
self.skills[meta.id] = meta
self.instructions[meta.id] = parts[2].strip()
logger.info("Registered skill [%s] from %s", meta.id, skill_path.name)
except Exception as e:
logger.warning("Failed to parse skill at %s: %s", skill_path, e)
def get_manifests_json(self) -> str:
return json.dumps([s.model_dump() for s in self.skills.values()], indent=2)
def get_instructions(self, skill_id: str) -> str:
return self.instructions.get(skill_id, "Execute standard task instructions accurately.")
# =====================================================================
# 4. Deterministic DAG Topological Sorter
# =====================================================================
class CyclicDependencyError(ValueError):
pass
class MissingDependencyError(KeyError):
pass
def sort_planned_tasks(tasks: List[PlannedTask]) -> List[PlannedTask]:
if not tasks:
return []
task_map: Dict[str, PlannedTask] = {t.task_id: t for t in tasks}
adjacency_list: Dict[str, List[str]] = defaultdict(list)
in_degree: Dict[str, int] = {t.task_id: 0 for t in tasks}
for task in tasks:
for dep_id in task.dependencies:
if dep_id not in task_map:
raise MissingDependencyError(f"Task '{task.task_id}' depends on undefined task '{dep_id}'.")
adjacency_list[dep_id].append(task.task_id)
in_degree[task.task_id] += 1
zero_in_degree = sorted([t_id for t_id, deg in in_degree.items() if deg == 0])
ready_queue: deque[str] = deque(zero_in_degree)
sorted_tasks: List[PlannedTask] = []
while ready_queue:
current_id = ready_queue.popleft()
sorted_tasks.append(task_map[current_id])
next_ready = []
for dependent_id in adjacency_list[current_id]:
in_degree[dependent_id] -= 1
if in_degree[dependent_id] == 0:
next_ready.append(dependent_id)
next_ready.sort()
ready_queue.extend(next_ready)
if len(sorted_tasks) != len(tasks):
unresolved = sorted([t_id for t_id, deg in in_degree.items() if deg > 0])
raise CyclicDependencyError(f"Cyclic dependency detected among tasks: {unresolved}")
return sorted_tasks
# =====================================================================
# 5. Deterministic Validator
# =====================================================================
class ArtifactSyntaxValidator:
@staticmethod
def validate(file_path: Path, content: str) -> Tuple[bool, Optional[str]]:
suffix = file_path.suffix.lower()
if suffix == ".py":
try:
ast.parse(content, filename=str(file_path))
return True, None
except SyntaxError as e:
return False, f"Python SyntaxError at line {e.lineno}, col {e.offset}: {e.msg}\nCode segment: {e.text}"
except Exception as e:
return False, f"Python AST Error: {str(e)}\n{traceback.format_exc()}"
elif suffix == ".json":
try:
json.loads(content)
return True, None
except json.JSONDecodeError as e:
return False, f"JSONDecodeError at line {e.lineno}, col {e.colno}: {e.msg}"
elif suffix == ".sql":
con = sqlite3.connect(":memory:")
try:
con.executescript(content)
return True, None
except sqlite3.Error as e:
return False, f"SQLite Execution Error: {str(e)}"
finally:
con.close()
return True, None
# =====================================================================
# 6. Stage 1: Planning Orchestrator
# =====================================================================
class Stage1Planner:
def __init__(self, client: LocalOllamaClient, registry: SkillLibraryRegistry, score_threshold: int = 7):
self.client = client
self.registry = registry
self.score_threshold = score_threshold
self.activity_log: List[Dict[str, Any]] = []
def _deterministic_check(self, spec: RefinedProblem) -> Tuple[bool, List[str]]:
violations = []
if len(spec.core_objective.split()) < 5:
violations.append("Core objective is too brief or ambiguous.")
if not spec.scope_in:
violations.append("scope_in must contain at least one concrete item.")
if not spec.scope_out:
violations.append("scope_out cannot be empty (negative boundaries missing).")
if not spec.identified_constraints:
violations.append("identified_constraints cannot be empty.")
return len(violations) == 0, violations
def refine_with_critic_loop(
self, raw_prompt: str, max_iterations: int = 3
) -> Tuple[RefinedProblem, CircuitBreakerStatus, List[int]]:
feedback_history: List[str] = []
candidate_spec: Optional[RefinedProblem] = None
score_trajectory: List[int] = []
refiner_instructions = self.registry.get_instructions("planning.problem_refiner")
for iteration in range(1, max_iterations + 1):
logger.info("Stage 1: Refinement Iteration %d/%d", iteration, max_iterations)
refiner_system = (
"You are an expert Systems Requirements & Problem Disambiguation Agent.\n"
"Restate the problem statement, isolate hard boundaries, and unearth hidden assumptions.\n"
f"{refiner_instructions}\n"
)
if feedback_history:
refiner_system += "\nPrevious Auditor Critiques to Resolve:\n" + "\n".join(f"- {f}" for f in feedback_history)
candidate_spec = self.client.generate_structured(
system_prompt=refiner_system,
user_prompt=raw_prompt,
schema_class=RefinedProblem,
)
valid_rules, violations = self._deterministic_check(candidate_spec)
if not valid_rules:
feedback_history.extend(violations)
score_trajectory.append(2)
self.activity_log.append({
"stage": "Stage 1 - Critic",
"iteration": iteration,
"status": "Deterministic Rule Failure",
"violations": violations,
"score": 2,
})
continue
critic_system = (
"You are a Systems Verification Auditor. Grade the candidate RefinedProblem on a 1-10 scale.\n"
"Evaluate scope clarity, completeness of negative constraints, and feasibility.\n"
f"A score of >= {self.score_threshold} means passed=True."
)
critic_user = f"Original Prompt:\n{raw_prompt}\n\nCandidate Spec:\n{candidate_spec.model_dump_json(indent=2)}"
critique: CriticEvaluationResult = self.client.generate_structured(
system_prompt=critic_system,
user_prompt=critic_user,
schema_class=CriticEvaluationResult,
)
score_trajectory.append(critique.overall_score)
self.activity_log.append({
"stage": "Stage 1 - Critic",
"iteration": iteration,
"score": critique.overall_score,
"passed": critique.passed,
"findings": critique.findings,
"feedback": critique.actionable_feedback,
})
if critique.passed and critique.overall_score >= self.score_threshold:
logger.info("Problem Refinement approved on iteration %d (Score: %d/10)", iteration, critique.overall_score)
return candidate_spec, CircuitBreakerStatus.CLOSED, score_trajectory
feedback_history.extend(critique.actionable_feedback)
# Stagnation Circuit Breaker Check
if len(score_trajectory) > 1 and score_trajectory[-1] <= score_trajectory[-2]:
logger.warning("Circuit Breaker Stagnation detected. Recovering via deterministic boundary patch.")
candidate_spec.scope_out.append("External cloud services outside local execution scope")
candidate_spec.identified_constraints.append("Must run deterministically using local Python runtime")
return candidate_spec, CircuitBreakerStatus.RECOVERED, score_trajectory
return candidate_spec, CircuitBreakerStatus.QUARANTINED, score_trajectory
def decompose(self, refined_spec: RefinedProblem) -> Stage1PlanningOutput:
logger.info("Stage 1: Decomposing refined spec and mapping skill catalog...")
catalog_manifests = self.registry.get_manifests_json()
system_prompt = (
"You are a Master Systems Architect & Task Decomposition Orchestrator.\n"
"Decompose the formal problem specification into sequential execution tasks.\n"
"Assign one or more matching skill IDs from the Available Skill Catalog.\n"
f"Available Skill Catalog:\n{catalog_manifests}"
)
user_prompt = f"Problem Specification:\n{refined_spec.model_dump_json(indent=2)}"
plan: Stage1PlanningOutput = self.client.generate_structured(
system_prompt=system_prompt,
user_prompt=user_prompt,
schema_class=Stage1PlanningOutput,
)
plan.refined_problem = refined_spec
return plan
# =====================================================================
# 7. Stage 2: Execution Orchestrator with Self-Correction & Diffing
# =====================================================================
class Stage2Executor:
def __init__(
self,
client: LocalOllamaClient,
registry: SkillLibraryRegistry,
output_workspace: Path,
max_predecessor_chars: int = 4000,
max_fix_retries: int = 3,
):
self.client = client
self.registry = registry
self.output_workspace = output_workspace
self.output_workspace.mkdir(parents=True, exist_ok=True)
self.max_predecessor_chars = max_predecessor_chars
self.max_fix_retries = max_fix_retries
self.results: List[TaskExecutionResult] = []
self.completed_artifacts: Dict[str, Path] = {}
def _clean_markdown_fences(self, content: str) -> str:
clean = content.strip()
if clean.startswith("```"):
lines = clean.splitlines()
if len(lines) >= 3 and lines[0].startswith("```") and lines[-1].startswith("```"):
clean = "\n".join(lines[1:-1])
return clean.strip()
def _compute_diff(self, old_text: str, new_text: str, filename: str) -> str:
old_lines = old_text.splitlines(keepends=True)
new_lines = new_text.splitlines(keepends=True)
diff = difflib.unified_diff(
old_lines,
new_lines,
fromfile=f"a/{filename} (faulty)",
tofile=f"b/{filename} (repaired)",
lineterm="",
)
return "".join(diff).strip()
def _build_predecessor_context(self, task: PlannedTask) -> str:
if not task.dependencies:
return ""
context_blocks = []
for dep_id in task.dependencies:
artifact_file = self.completed_artifacts.get(dep_id)
if not artifact_file or not artifact_file.exists():
continue
try:
raw_text = artifact_file.read_text(encoding="utf-8")
truncated = (
raw_text
if len(raw_text) <= self.max_predecessor_chars
else raw_text[: self.max_predecessor_chars // 2]
+ "\n... [TRUNCATED] ...\n"
+ raw_text[-self.max_predecessor_chars // 2 :]
)
rel_path = artifact_file.relative_to(self.output_workspace)
context_blocks.append(
f"#### Upstream Artifact from [{dep_id}] (`{rel_path}`):\n```text\n{truncated}\n```"
)
except Exception as err:
logger.error("Failed reading upstream artifact %s: %s", artifact_file, err)
if not context_blocks:
return ""
return (
"\n### UPSTREAM ARTIFACT CONTEXT (Predecessor Dependencies):\n"
"Align your code strictly with types, classes, functions, and schemas defined below:\n\n"
+ "\n\n".join(context_blocks)
+ "\n"
)
def _generate_remedy(
self,
task: PlannedTask,
assigned_skill_id: str,
skill_instructions: str,
flawed_content: str,
error_message: str,
attempt: int,
) -> str:
logger.info("Executing self-correction attempt %d/%d for [%s]...", attempt, self.max_fix_retries, task.task_id)
system_prompt = (
f"You are an Expert Automated Code Repair Agent using skill [{assigned_skill_id}].\n"
f"Target File: {task.expected_artifact_path}\n"
f"{skill_instructions}\n\n"
"Your previous code generation attempt failed validation with a syntax or execution error.\n"
"Inspect the error traceback and prior broken code, resolve the root cause, and return the complete corrected file.\n"
"Return ONLY the raw corrected file contents. Do not truncate sections or omit existing working logic."
)
user_prompt = (
f"### ERROR ENCOUNTERED:\n{error_message}\n\n"
f"### PRIOR FAULTY FILE CONTENT:\n```text\n{flawed_content}\n```\n\n"
"Provide the complete, functional, fixed file content resolving the error above."
)
remedy_raw = self.client.generate_text(system_prompt, user_prompt)
return self._clean_markdown_fences(remedy_raw)
def execute_plan(self, plan: Stage1PlanningOutput) -> List[TaskExecutionResult]:
logger.info("Topologically sorting %d planned tasks...", len(plan.execution_plan))
ordered_tasks = sort_planned_tasks(plan.execution_plan)
for task in ordered_tasks:
t_start = time.perf_counter()
logger.info("Executing [%s]: %s", task.task_id, task.title)
assigned_skill_id = task.recommended_skills[0] if task.recommended_skills else "general"
skill_instructions = self.registry.get_instructions(assigned_skill_id)
predecessor_context = self._build_predecessor_context(task)
agent_system_prompt = (
f"You are a Specialized Execution Agent powered by the [{assigned_skill_id}] skill.\n"
f"Task: {task.title}\n"
f"Description: {task.description}\n"
f"Target File: {task.expected_artifact_path}\n\n"
f"### Skill Operational Instructions:\n{skill_instructions}\n\n"
"Produce the complete, functional, production-ready file content without truncation."
)
agent_user_prompt = (
f"Generate the exact file artifact content for '{task.expected_artifact_path}'.\n\n"
f"### System Specification:\n"
f"- Objective: {plan.refined_problem.core_objective}\n"
f"- Constraints: {', '.join(plan.refined_problem.identified_constraints)}\n"
f"{predecessor_context}\n"
"Respond ONLY with the complete file contents to be written to disk."
)
generated_content = self.client.generate_text(agent_system_prompt, agent_user_prompt)
current_content = self._clean_markdown_fences(generated_content)
artifact_file = self.output_workspace / task.expected_artifact_path
artifact_file.parent.mkdir(parents=True, exist_ok=True)
is_valid, error_msg = ArtifactSyntaxValidator.validate(artifact_file, current_content)
correction_history: List[CorrectionEvent] = []
retry_count = 0
while not is_valid and retry_count < self.max_fix_retries:
retry_count += 1
prior_flawed = current_content
logger.warning(
"Task [%s] failed validation on attempt %d: %s",
task.task_id,
retry_count,
error_msg.splitlines()[0] if error_msg else "Unknown",
)
current_content = self._generate_remedy(
task=task,
assigned_skill_id=assigned_skill_id,
skill_instructions=skill_instructions,
flawed_content=prior_flawed,
error_message=error_msg or "Validation Failure",
attempt=retry_count,
)
diff_text = self._compute_diff(
old_text=prior_flawed,
new_text=current_content,
filename=task.expected_artifact_path,
)
correction_history.append(
CorrectionEvent(
attempt=retry_count,
error_message=error_msg or "Validation Failure",
pre_fix_snippet=prior_flawed[:600],
post_fix_snippet=current_content[:600],
unified_diff=diff_text,
timestamp_utc=datetime.now(timezone.utc).isoformat(),
)
)
is_valid, error_msg = ArtifactSyntaxValidator.validate(artifact_file, current_content)
if not is_valid:
logger.critical("Task [%s] failed all %d repair attempts.", task.task_id, self.max_fix_retries)
raise RuntimeError(f"Halted: Task [{task.task_id}] failed self-correction.\nError: {error_msg}")
artifact_file.write_text(current_content, encoding="utf-8")
self.completed_artifacts[task.task_id] = artifact_file
elapsed = round(time.perf_counter() - t_start, 2)
self.results.append(
TaskExecutionResult(
task_id=task.task_id,
title=task.title,
skill_used=assigned_skill_id,
artifact_path=str(artifact_file),
generated_content=current_content,
success=True,
runtime_sec=elapsed,
correction_history=correction_history,
)
)
return self.results
# =====================================================================
# 8. Markdown Audit Reporter
# =====================================================================
class MarkdownAuditReporter:
@staticmethod
def generate_report(
raw_prompt: str,
refined_spec: RefinedProblem,
circuit_status: CircuitBreakerStatus,
score_trajectory: List[int],
stage1_logs: List[Dict[str, Any]],
plan: Stage1PlanningOutput,
execution_results: List[TaskExecutionResult],
report_dir: Path,
) -> Path:
report_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
report_file = report_dir / f"orchestration_run_{timestamp}.md"
total_corrections = sum(len(res.correction_history) for res in execution_results)
trajectory_str = " -> ".join(f"**{s}/10**" for s in score_trajectory)
md = f"""# Multi-Agent Workflow Execution & Audit Report
- **Run Timestamp:** `{datetime.now(timezone.utc).isoformat()}`
- **Circuit Breaker Status:** `{circuit_status.value.upper()}`
- **Critic Score Progression:** {trajectory_str}
- **Stage 2 Self-Correction Events:** `{total_corrections}`
- **Artifacts Delivered:** `{len(execution_results)}`
---
## 1. Problem Specification & Scope Boundaries
### Raw User Prompt
> {raw_prompt}
### Refined Core Objective
> {refined_spec.core_objective}
| Scope In | Scope Out | Identified Constraints | Key Assumptions |
| :--- | :--- | :--- | :--- |
| {chr(10).join(f"- {s}" for s in refined_spec.scope_in)} | {chr(10).join(f"- {s}" for s in refined_spec.scope_out)} | {chr(10).join(f"- {s}" for s in refined_spec.identified_constraints)} | {chr(10).join(f"- {s}" for s in refined_spec.key_assumptions)} |
---
## 2. Stage 1 Verification & Critic Evaluation Trail
"""
for log in stage1_logs:
md += f"### Iteration {log.get('iteration')} Audit\n"
md += f"- **Score:** `{log.get('score')}/10` (Passed: `{log.get('passed', False)}`)\n"
if log.get("findings"):
md += f"- **Findings:** {log.get('findings')}\n"
if log.get("feedback"):
md += "- **Actionable Feedback:**\n"
for fb in log["feedback"]:
md += f" - {fb}\n"
if log.get("violations"):
md += f"- **Deterministic Violations:** {', '.join(log['violations'])}\n"
md += "\n"
md += """---
## 3. Stage 1 Execution Plan (DAG)
| Task ID | Task Title | Assigned Skill | Dependencies | Target File |
| :--- | :--- | :--- | :--- | :--- |
"""
for t in plan.execution_plan:
skills = ", ".join(t.recommended_skills) or "general"
deps = ", ".join(t.dependencies) or "None"
md += f"| `{t.task_id}` | {t.title} | `{skills}` | {deps} | `{t.expected_artifact_path}` |\n"
md += """
---
## 4. Stage 2 Execution Results
| Task ID | Status | Hydrated Skill | Latency | Self-Corrections | Target Artifact |
| :--- | :--- | :--- | :--- | :--- | :--- |
"""
for res in execution_results:
status_badge = "✅ Passed" if res.success else "❌ Failed"
corrections_count = len(res.correction_history)
md += f"| `{res.task_id}` | {status_badge} | `{res.skill_used}` | {res.runtime_sec}s | {corrections_count} | `{res.artifact_path}` |\n"
md += """
---
## 5. Self-Correction & Automated Repair Trail
"""
if total_corrections == 0:
md += "_No syntax or execution errors encountered. All artifacts passed initial verification._\n\n"
else:
for res in execution_results:
if not res.correction_history:
continue
md += f"### Task [{res.task_id}]: {res.title}\n"
md += f"- **Artifact Path:** `{res.artifact_path}`\n"
md += f"- **Total Repair Passes:** `{len(res.correction_history)}`\n\n"
for event in res.correction_history:
md += f"#### Attempt #{event.attempt} Remediation Trace (`{event.timestamp_utc}`)\n\n"
md += "**Triggering Traceback:**\n```text\n" + event.error_message.strip() + "\n```\n\n"
md += "**Applied Code Diff:**\n```diff\n" + (event.unified_diff if event.unified_diff else "# No lines changed") + "\n```\n\n"
md += """---
## 6. Generated Artifact Code Listings
"""
for res in execution_results:
md += f"### `{res.artifact_path}`\n"
md += f"```python\n{res.generated_content.strip()}\n```\n\n"
report_file.write_text(md, encoding="utf-8")
return report_file
# =====================================================================
# 9. Skill Library Bootstrapper
# =====================================================================
def bootstrap_skill_library(skills_dir: Path):
if skills_dir.exists():
return
# Skill 1: Planning / Problem Refiner
refiner_dir = skills_dir / "planning" / "problem_refiner"
refiner_dir.mkdir(parents=True, exist_ok=True)
(refiner_dir / "SKILL.md").write_text(
"""---
id: "planning.problem_refiner"
name: "Problem Refiner"
version: "1.0.0"
stage: "planning"
description: "Disambiguates scope boundaries and extracts assumptions."
tags: ["refinement", "requirements"]
---
Analyze problem statements to isolate core functional goals, negative boundaries, and unstated assumptions.
""",
encoding="utf-8",
)
# Skill 2: Implementation / Python Scaffolder
py_dir = skills_dir / "implementation" / "python_scaffolder"
py_dir.mkdir(parents=True, exist_ok=True)
(py_dir / "SKILL.md").write_text(
"""---
id: "implementation.python_scaffolder"
name: "Python Scaffolder"
version: "1.0.0"
stage: "implementation"
description: "Generates robust, typed Python modules and data structures."
tags: ["python", "codegen"]
---
Generate modular, typed Python code with complete function bodies, docstrings, and robust error handling.
""",
encoding="utf-8",
)
# Skill 3: Implementation / SQLite Builder
sql_dir = skills_dir / "implementation" / "sqlite_builder"
sql_dir.mkdir(parents=True, exist_ok=True)
(sql_dir / "SKILL.md").write_text(
"""---
id: "implementation.sqlite_builder"
name: "SQLite Schema Builder"
version: "1.0.0"
stage: "implementation"
description: "Designs relational SQLite tables, schema migrations, and DDL scripts."
tags: ["sql", "sqlite", "database"]
---
Generate valid, idempotent SQL DDL scripts with primary keys, indexes, and valid column constraints.
""",
encoding="utf-8",
)
# =====================================================================
# 10. Application Entrypoint
# =====================================================================
def run_application(
user_problem: str,
model_name: str = "llama3.1:latest",
run_dir: Path = Path("./orchestrator_run"),
):
skills_dir = run_dir / "skills_library"
workspace_dir = run_dir / "workspace"
reports_dir = run_dir / "reports"
# Step 1: Initialize local environment & discover skills
bootstrap_skill_library(skills_dir)
registry = SkillLibraryRegistry(skills_dir)
registry.scan()
ollama_client = LocalOllamaClient(model_name=model_name)
# Step 2: Stage 1 Planning (Refinement + Critic Loop + Circuit Breaker)
planner = Stage1Planner(client=ollama_client, registry=registry, score_threshold=7)
refined_spec, status, score_trajectory = planner.refine_with_critic_loop(user_problem, max_iterations=3)
plan = planner.decompose(refined_spec)
# Step 3: Stage 2 Execution (Topological Ordering + Self-Correction)
executor = Stage2Executor(client=ollama_client, registry=registry, output_workspace=workspace_dir)
execution_results = executor.execute_plan(plan)
# Step 4: Markdown Audit Export
report_file = MarkdownAuditReporter.generate_report(
raw_prompt=user_problem,
refined_spec=refined_spec,
circuit_status=status,
score_trajectory=score_trajectory,
stage1_logs=planner.activity_log,
plan=plan,
execution_results=execution_results,
report_dir=reports_dir,
)
print("\n" + "=" * 65)
print("ORCHESTRATION PIPELINE COMPLETE")
print("=" * 65)
print(f"Status : {status.value.upper()}")
print(f"Refined Objective : {refined_spec.core_objective}")
print(f"Tasks Delivered : {len(execution_results)}")
print(f"Workspace Directory : {workspace_dir.resolve()}")
print(f"Markdown Audit Report: {report_file.resolve()}")
if __name__ == "__main__":
sample_problem = (
"Build a directory ingest watcher that checks an inbox folder for new JSON telemetry payload files, "
"validates them against a schema, and saves valid records into an SQLite database with error quarantine handling."
)
# Execute against local Ollama instance (ensure `ollama serve` is active)
model = sys.argv[1] if len(sys.argv) > 1 else "llama3.1:latest"
run_application(user_problem=sample_problem, model_name=model)
Running the Application
Make sure Ollama is running locally with your desired weights loaded:
ollama pull llama3.1Install the required client dependencies:
pip install ollama pydantic pyyamlRun the script:
python app.py llama3.1:latest
All generated code artifacts are saved to ./orchestrator_run/workspace/, and the complete execution trail—including the Evaluator-Critic scoring trajectory, DAG task tables, error tracebacks, and git diffs—is written to ./orchestrator_run/reports/orchestration_run_<timestamp>.md.
From <https://gemini.google.com/app/ccd78610e4ce0264> Google Gemini (3.8 Flash)
Comments
Post a Comment