Skip to content

Recursive Architect

recursive_architect -- G6 Recursive Architect Meta-Plugin (DEPRECATED).

Cluster: Goal & Planning | Type: component | MCP Tools: None

Overview

Recursive architect that orchestrates the full G6 hyperdistillation pipeline through a 7-tuple specification: (goal, resources, context, constraints, breakpoints, guardrails, checkpoints). Decomposes goals into a branching search tree, evaluates each branch under CSF safety constraints, supports HITL breakpoints, captures git snapshots, and writes crash logs — reducing LLM agency progressively as solutions prove out.

Deprecation notice:

Recursive Architect is deprecated for new core solver work. Use mvp.reliability_pipeline.ReliabilityPipelineBlock for new solver implementations. Recursive Architect remains available for legacy orchestration, HITL, blueprint, and diagnostic workflows; MCP responses include a warning_card with reason: deprecated_core_solver and the successor block.

When to use:

  • Decomposing a high-level engineering goal into a verified, constrained, multi-branch execution plan
  • Running the full G6 pipeline (decompose → branch → CSF-check → execute → checkpoint → snapshot) as a single call
  • Enforcing resource budgets, hard constraints, and HITL approval gates across a complex multi-step task

Durable execution warning:

Recursive Architect MCP submission uses adapt_rq as the preferred durable execution path when Redis/RQ is available. True restart-resumable execution requires both Redis and an RQ worker listening on the g6-architect queue. If either is missing, the MCP server falls back to local in-process execution; that fallback is useful for first-run local workflows, but it will not resume automatically after the MCP process exits or restarts.

Start a worker with:

python -m rq worker g6-architect --url redis://localhost:6379/0

Useful environment variables:

  • REDIS_URL or G6_REDIS_URL sets the Redis connection.
  • G6_ARCHITECT_QUEUE changes the queue name from g6-architect.
  • G6_ARCHITECT_DISABLE_RQ=1 forces local fallback for debugging.

Capability discovery and envelopes:

  • arch_capabilities reports backing modules, optional infrastructure, execution backend, production readiness, deprecation, and self-modification gate status without starting a run.
  • Every MCP tool returns the canonical envelope fields completion_state, warning_card, evidence, request_id, and data.
  • Legal completion_state values are exactly verified, qualified-draft, and blocked-escalated.
  • ENABLE_SELF_MODIFICATION defaults to disabled. MetaProgrammingBlock and CEGISBlock are gated; production self-model checks fail closed, while non-production self-model checks fail open but still require the environment flag for code-modifying components.

Example:

from mvp.recursive_architect import RecursiveArchitectBlock, ArchitectInput

block = RecursiveArchitectBlock(name="architect")
result = block.infer(ArchitectInput(
    goal="Implement a REST endpoint for user registration",
    context="FastAPI project, PostgreSQL backend",
    workspace_root="/workspace/myproject",
))
# result.ok → True; result.value → ArchitectOutput with completed branches, metrics

Works well with: autonomous_orchestrator, solver, hyperdistillation

Public API

RecursiveArchitectBlock(AIBlock[ArchitectInput, ArchitectOutput, ArchitectState])

OS-level orchestrator for the G6 hyperdistillation system.

Field Type Default
name str 'recursive_architect'
resource_bounds ResourceBounds \| None None
metrics_port int 9092

Methods:

infer(data: ArchitectInput) -> Result[ArchitectOutput]

Synchronous entry point. Wraps asyncio.run() around run_async().

run_async(data: ArchitectInput, _ctx = None) -> ArchitectOutput

Async version for use inside Textual's event loop.

pause() -> None

Pause execution (clears play_event).

resume() -> None

Resume execution (sets play_event).

stop() -> None

Request graceful stop (sets stop_event).

AgenticPattern

A single vendored agentic pattern.

Field Type Default
slug str required
title str required
status str required
category str required
source str required
authors tuple[str, ...] ()
tags tuple[str, ...] ()
summary str ''
maturity str ''
complexity str ''
effort str ''
impact str ''
signals tuple[str, ...] ()
anti_signals tuple[str, ...] ()
prerequisites tuple[str, ...] ()
related tuple[str, ...] ()
anti_patterns tuple[str, ...] ()
tools tuple[str, ...] ()
domains tuple[str, ...] ()
updated_at str ''
path str ''
sections dict[str, str] field(default_factory=dict)

Methods:

compact(include_sections: bool = False) -> dict[str, Any]

PatternCatalog

Read-only index over the vendored agentic pattern corpus.

Constructor:

Parameter Type Default
patterns_dir Path PATTERNS_DIR

Methods:

list_patterns() -> list[AgenticPattern]

get(slug: str) -> AgenticPattern | None

manifest() -> dict[str, Any]

search(query: str = '', category: str = '', tags: list[str] | None = None, limit: int = 20) -> list[AgenticPattern]

candidate_context(query: str, limit: int = 18) -> str

ResourceBoundsSpec(BaseModel)

Resource budget for a recursive architect run.

Field Type Default
max_cost_usd float \| None None
max_tokens_per_minute int \| None None
max_execution_seconds float \| None None

Methods:

non_negative(v: Any) -> Any

ConstraintSpec(BaseModel)

A single hard constraint to enforce during execution.

Field Type Default
raw str required
target_branch str \| None None

BreakpointSpec(BaseModel)

A HITL breakpoint definition.

Field Type Default
name str required
active bool True
trigger_on str 'any'
timeout_seconds float 300.0

GuardrailSpec(BaseModel)

A soft guardrail to check before each branch.

Field Type Default
name str required
expression str required
message str ''

CheckpointSpec(BaseModel)

A numeric threshold checkpoint.

Field Type Default
name str required
description str ''
metric_key str ''
threshold float 0.0

ArchitectInput(BaseModel)

The full 7-tuple specification for a Recursive Architect run.

Field Type Default
goal str required
resources ResourceBoundsSpec Field(default_factory=ResourceBoundsSpec)
context str ''
constraints list[ConstraintSpec] Field(default_factory=list)
breakpoints list[BreakpointSpec] Field(default_factory=list)
guardrails list[GuardrailSpec] Field(default_factory=list)
checkpoints list[CheckpointSpec] Field(default_factory=list)
subtasks list[str] Field(default_factory=list)
blueprint dict[str, Any] Field(default_factory=dict)
run_id str ''
workspace_base_dir str ''
max_depth int 3
max_branches int 50

Methods:

goal_not_empty(v: str) -> str

max_depth_non_negative(v: int) -> int

max_branches_positive(v: int) -> int

model_post_init(__context: Any) -> None

NodeStatus(str, Enum)

HumanTask

A task emitted to the HITL queue for human review.

Field Type Default
task_id str required
run_id str required
breakpoint_name str required
node_id str required
node_goal str required
branch_depth int required
trigger_reason str required
created_at str field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
context_summary str ''
pending_branches list[str] field(default_factory=list)
timeout_seconds float 300.0
metacognitive_prompt str ''
prompt_type str ''
prompt_options list[str] field(default_factory=list)
skill_phase str ''
is_challenge bool False
human_prediction str ''
human_confidence float 0.0
override_reason str ''
approval_action dict[str, Any] \| None None
rejection_action dict[str, Any] \| None None
decision_effect str ''
escalated_task_id str ''
escalated_at str ''

Methods:

task_filename() -> str

done_filename() -> str

BranchResult

Result from executing a single SearchTree branch.

Field Type Default
node_id str required
node_goal str required
status NodeStatus NodeStatus.PENDING
component_used str ''
output_summary str ''
error str ''
commit_sha str ''
elapsed_seconds float 0.0
csf_vub float 0.0
tokens_used int 0
patterns_used list[str] field(default_factory=list)
blueprint_id str ''
agent_role str ''
verification_gate str ''
pattern_selection_rationale str ''
llm_decision_trace_refs list[str] field(default_factory=list)
skill_context_pack dict[str, Any] field(default_factory=dict)

ArchitectState

Mutable runtime state for the RecursiveArchitectBlock.

Field Type Default
run_id str ''
workspace_root str ''
branch_results list[BranchResult] field(default_factory=list)
pending_hitl list[HumanTask] field(default_factory=list)
log_lines list[str] field(default_factory=list)
metrics dict[str, float] field(default_factory=dict)
checkpoint_log list[dict] field(default_factory=list)
git_snapshots list[str] field(default_factory=list)
stop_requested bool False
override_log list[dict] field(default_factory=list)
reflect_log list[dict] field(default_factory=list)
skill_phase_at_start str ''
process_blueprint dict[str, Any] field(default_factory=dict)
pattern_selection dict[str, Any] field(default_factory=dict)
llm_decision_log list[dict[str, Any]] field(default_factory=list)

ArchitectOutput(BaseModel)

Final output from a RecursiveArchitectBlock run.

Field Type Default
run_id str required
workspace_root str required
status str required
summary str required
branch_results list[dict] Field(default_factory=list)
pending_hitl list[dict] Field(default_factory=list)
git_snapshots list[str] Field(default_factory=list)
metrics dict[str, float] Field(default_factory=dict)
errors list[str] Field(default_factory=list)
degraded bool False
degradation_reason str \| None None
patterns_used list[str] Field(default_factory=list)
blueprint_id str ''
agent_roles list[str] Field(default_factory=list)
verification_gates list[str] Field(default_factory=list)
pattern_selection_rationale str ''
agent_trace_refs list[str] Field(default_factory=list)
reliability_envelope dict \| None None
agentic_evidence dict[str, Any] Field(default_factory=dict)

ArchitectSkill

Compact metadata for a local Codex skill.

Field Type Default
name str required
description str ''
source str ''
relative_path str ''
capabilities tuple[str, ...] ()
triggers tuple[str, ...] ()
risk_notes tuple[str, ...] ()
path str field(default='', repr=False)

Methods:

compact() -> dict[str, Any]

SkillCatalog

Read-only index over local Codex skills.

Constructor:

Parameter Type Default
roots dict[str, Path] \| None None

Methods:

list_skills() -> list[ArchitectSkill]

get(name: str) -> ArchitectSkill | None

search(query: str = '', source: str = '', limit: int = 20) -> list[ArchitectSkill]

Functions

get_pattern_catalog() -> PatternCatalog

get_skill_catalog() -> SkillCatalog