Cog Arch Soar¶

cog_arch_soar — SOAR cognitive architecture.

Cluster: Experience & Autonomy | Type: component | MCP Tools: 26

Overview¶

MVP-simplified SOAR (State, Operator And Result) cognitive architecture implementing a working memory, production rule system, and impasse-aware control loop. Delegates 26 operations to an inner SOARMCPBlock with SQLite-backed working memory and a rule evolver that learns new productions from resolved impasses, with optional LLM rule suggestion surfaced as degraded or unavailable when it cannot run.

When to use:

Implementing goal-directed symbolic reasoning with automatic sub-goaling when an impasse is reached
Storing and evolving a production rule library that grows from agent experience
Running SOAR-style cognitive cycles within the G6 pipeline alongside retrieval and planning components

Production caveat:

This component is suitable for pilot and MVP workflows where the goal is transparent symbolic state, rule tracing, and incremental learning. It should not be described as a complete SOAR implementation or compliance-grade reasoning system. Preference handling is intentionally simplified, real SOAR substates/backtracing are incomplete, some cross-component integrations can degrade to local fallbacks or stubs, and production claims should surface degraded, degradation_reason, completion_state, warning_card, evidence, and unavailable_backends when they appear in outputs.

Example:

from mvp.cog_arch_soar import CogArchSOARBlock, SOARInput

block = CogArchSOARBlock(name="soar")
block.infer(SOARInput(op="wm_add", identifier="s1", attribute="type", value="problem"))
result = block.infer(SOARInput(op="cycle_run", goal_test_json='{"identifier":"s1","attribute":"type","value":"problem"}'))
# result.ok -> True; result.value -> SOAROutput with elements, rules, message

Works well with: cog_arch_actr, cog_arch_gps, experience_loop

Public API¶

`StepTrace`¶

Trace of a single step execution within a solve.

Field	Type	Default
`step_num`	`int`	required
`status`	`str`	required
`operator_name`	`str`	required
`precondition_check`	`dict`	required
`postcondition_check`	`dict`	required
`cross_validation`	`dict`	required
`output_summary`	`str`	required
`tokens`	`int`	required
`latency_ms`	`int`	required

`Case`¶

A complete solve trace — the unit of learning.

Field	Type	Default
`case_id`	`str`	required
`problem_embedding`	`list[float]`	required
`goal`	`str`	required
`context`	`str`	required
`complexity`	`str`	required
`steps`	`list[StepTrace]`	required
`final_answer`	`str`	required
`success`	`bool`	required
`total_steps_run`	`int`	required
`total_tokens`	`int`	required
`chunks_fired`	`list[str]`	required
`impasses_hit`	`list[dict]`	required
`rules_snapshot`	`list[dict]`	required

`CaseBank`¶

SQLite + FAISS case storage for SOAR learning.

Constructor:

Parameter	Type	Default
`db_path`	`str`	required

Methods:

`store(case: Case) -> None`¶

Persist a case to SQLite and update FAISS index.

`store_validated(case: Case, trusted: bool = False) -> CaseIngestVerdict`¶

Store a case after validating its success claim (OWASP LLM04 ingest gate).

`retrieve(query_embedding: list[float], top_k: int = 8) -> list[Case]`¶

Retrieve most similar cases by embedding cosine similarity.

`count() -> int`¶

`get_recent(n: int) -> list[Case]`¶

`record_training_signal(query_embedding: list[float], retrieved_case_ids: list[str], success: bool) -> None`¶

Store (query, case, helpful) tuples for future retriever training.

`close() -> None`¶

`CaseRetriever`¶

Embed problems and retrieve similar past cases.

Methods:

`embed(goal: str, context: str | None = None) -> list[float]`¶

Embed a problem description into a 384-dim vector.

`retrieve_from_bank(case_bank: CaseBank, query_embedding: list[float], top_k: int = 8) -> tuple[list[Case], list[float]]`¶

Retrieve similar cases from case bank with similarity scores.

`inject_into_wm(cases: list[Case], wm: WorkingMemory, similarity_scores: list[float] | None = None) -> None`¶

Inject retrieved cases into Working Memory as WMEs.

`StepContextEnricher`¶

Load the step goal and ctx_ace observations into Working Memory.

Methods:

`enrich(step_num: int, project_state: dict, wm: WorkingMemory) -> None`¶

Enrich WM with the step goal context and prior-step observations.

`RuleEvolver`¶

Evolve SOAR operator rules using evoskill feedback descent.

Constructor:

Parameter	Type	Default
`safety_threshold`	`float`	0.5

Methods:

`evolve(current_rules: list[dict], case_bank: CaseBank, k_recent: int = 10) -> list[dict]`¶

One evolution cycle on the rule set.

`SOARInput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`identifier`	`str`	''
`attribute`	`str`	''
`value`	`str`	''
`wme_id`	`str`	''
`wme_type`	`str`	'state'
`depth`	`int`	0
`rule_id`	`str`	''
`rule_name`	`str`	''
`conditions_json`	`str`	'[]'
`actions_json`	`str`	'[]'
`rule_description`	`str`	''
`max_cycles`	`int`	100
`goal_test_json`	`str`	''
`operator_name`	`str`	''
`preferences_json`	`str`	'["acceptable"]'
`operator_actions_json`	`str`	'[]'
`impasse_id`	`str`	''
`impasse_type`	`str`	''
`chunk_id`	`str`	''
`chunk_conditions_json`	`str`	'[]'
`chunk_actions_json`	`str`	'[]'
`prompt`	`str`	''
`system_prompt`	`str`	''
`description`	`str`	''
`wm_context_json`	`str`	'[]'
`context_json`	`str`	'{}'
`bridge_payload_json`	`str`	'{}'
`hat_action`	`str`	''
`query`	`str`	''
`top_k`	`int`	10
`limit`	`int`	50
`session_id`	`str`	''

`SOAROutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`ok`	`bool`	required
`elements`	`list[dict]`	Field(default_factory=list)
`rules`	`list[dict]`	Field(default_factory=list)
`message`	`str`	''
`error`	`str`	''
`data_json`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`unavailable_backends`	`list[str]`	Field(default_factory=list)

`CogArchSOARBlock(AIBlock[SOARInput, SOAROutput, dict])`¶

SOAR cognitive architecture block.

Field	Type	Default
`name`	`str`	'cog_arch_soar'
`state`	`dict \\| None`	None
`resource_bounds`	`Any`	None

Methods:

`infer(data: SOARInput) -> Result[SOAROutput]`¶

`SoarSolverBridge`¶

SOAR meta-controller over the 15-step solver.

Constructor:

Parameter	Type	Default
`solver_input`	`SolverInput`	required
`project_dir`	`str`	required
`llm_backend`	`str`	required
`registry`	`-`	None
`chunk_library`	`list[dict] \\| None`	None
`max_cycles`	`int`	100
`max_retries_per_step`	`int`	3
`case_bank`	`CaseBank \\| None`	None
`enable_verification`	`bool`	True
`enable_retrieval`	`bool`	True
`enable_evolution`	`bool`	True
`evolution_k`	`int`	10

Methods:

`solve() -> SolverOutput`¶

Run the SOAR decision cycle with deep integration hooks.

`get_chunks() -> list[dict]`¶

Return the current chunk library for persistence.

`GateAction(Enum)`¶

`GateResult`¶

Field	Type	Default
`action`	`GateAction`	required
`contract`	`dict`	required
`issues`	`list[str]`	field(default_factory=list)
`unavailable_backends`	`list[str]`	field(default_factory=list)

`VerificationResult`¶

Field	Type	Default
`verified`	`bool`	required
`verifier_used`	`str`	required
`proof`	`dict`	field(default_factory=dict)
`cross_checks`	`list[dict]`	field(default_factory=list)
`issues`	`list[str]`	field(default_factory=list)
`unavailable_backends`	`list[str]`	field(default_factory=list)
`status`	`str`	SKIPPED

`StepVerifier`¶

Pre-step introspection + post-step formal verification.

Methods:

`introspect_and_gate(step_num: int, project_state: dict) -> GateResult`¶

Pre-step: introspect component contracts and gate on preconditions.

`verify_and_cross_check(step_num: int, step_output: dict, project_state: dict) -> VerificationResult`¶

Post-step: verify postconditions + cross-validate.

`verify_step(step_index: int, step_output: dict, context: dict) -> dict`¶

Verify a step and accurately report which tools ran vs were skipped.

Functions¶

`build_operator_rules() -> list[dict]`¶

Convert STEP_REGISTRY into SOAR production rules.

`mirror_to_wm(wm: WorkingMemory, step_num: int, status: str, output: dict) -> None`¶

Mirror step results into Working Memory as WMEs.

MCP Tools¶

Operation	Source
`wm_add`	cog_arch_soar_mcp
`wm_get`	cog_arch_soar_mcp
`wm_remove`	cog_arch_soar_mcp
`wm_list`	cog_arch_soar_mcp
`rule_create`	cog_arch_soar_mcp
`rule_get`	cog_arch_soar_mcp
`rule_list`	cog_arch_soar_mcp
`rule_delete`	cog_arch_soar_mcp
`cycle_run`	cog_arch_soar_mcp
`cycle_step`	cog_arch_soar_mcp
`cycle_status`	cog_arch_soar_mcp
`op_propose`	cog_arch_soar_mcp
`op_select`	cog_arch_soar_mcp
`op_apply`	cog_arch_soar_mcp
`impasse_detect`	cog_arch_soar_mcp
`impasse_resolve_llm`	cog_arch_soar_mcp
`impasse_list`	cog_arch_soar_mcp
`chunk_learn`	cog_arch_soar_mcp
`chunk_list`	cog_arch_soar_mcp
`llm_suggest_rule`	cog_arch_soar_mcp
`find_analogy`	cog_arch_soar_mcp
`realtime_bridge`	cog_arch_soar_mcp
`hat_gate`	cog_arch_soar_mcp
`soar_search`	cog_arch_soar_mcp
`soar_info`	cog_arch_soar_mcp
`list_patterns`	cog_arch_soar_mcp

Cog Arch Soar¶

Overview¶

Public API¶

StepTrace¶

Case¶

CaseBank¶

store(case: Case) -> None¶

store_validated(case: Case, trusted: bool = False) -> CaseIngestVerdict¶

retrieve(query_embedding: list[float], top_k: int = 8) -> list[Case]¶

count() -> int¶

get_recent(n: int) -> list[Case]¶

record_training_signal(query_embedding: list[float], retrieved_case_ids: list[str], success: bool) -> None¶

close() -> None¶

CaseRetriever¶

embed(goal: str, context: str | None = None) -> list[float]¶

retrieve_from_bank(case_bank: CaseBank, query_embedding: list[float], top_k: int = 8) -> tuple[list[Case], list[float]]¶

inject_into_wm(cases: list[Case], wm: WorkingMemory, similarity_scores: list[float] | None = None) -> None¶

StepContextEnricher¶

enrich(step_num: int, project_state: dict, wm: WorkingMemory) -> None¶

RuleEvolver¶

evolve(current_rules: list[dict], case_bank: CaseBank, k_recent: int = 10) -> list[dict]¶

SOARInput(BaseModel)¶

SOAROutput(BaseModel)¶

CogArchSOARBlock(AIBlock[SOARInput, SOAROutput, dict])¶

infer(data: SOARInput) -> Result[SOAROutput]¶

SoarSolverBridge¶

solve() -> SolverOutput¶

get_chunks() -> list[dict]¶

GateAction(Enum)¶

GateResult¶

VerificationResult¶

StepVerifier¶

introspect_and_gate(step_num: int, project_state: dict) -> GateResult¶

verify_and_cross_check(step_num: int, step_output: dict, project_state: dict) -> VerificationResult¶

verify_step(step_index: int, step_output: dict, context: dict) -> dict¶

Functions¶

build_operator_rules() -> list[dict]¶

mirror_to_wm(wm: WorkingMemory, step_num: int, status: str, output: dict) -> None¶

MCP Tools¶

`StepTrace`¶

`Case`¶

`CaseBank`¶

`store(case: Case) -> None`¶

`store_validated(case: Case, trusted: bool = False) -> CaseIngestVerdict`¶

`retrieve(query_embedding: list[float], top_k: int = 8) -> list[Case]`¶

`count() -> int`¶

`get_recent(n: int) -> list[Case]`¶

`record_training_signal(query_embedding: list[float], retrieved_case_ids: list[str], success: bool) -> None`¶

`close() -> None`¶

`CaseRetriever`¶

`embed(goal: str, context: str | None = None) -> list[float]`¶

`retrieve_from_bank(case_bank: CaseBank, query_embedding: list[float], top_k: int = 8) -> tuple[list[Case], list[float]]`¶

`inject_into_wm(cases: list[Case], wm: WorkingMemory, similarity_scores: list[float] | None = None) -> None`¶

`StepContextEnricher`¶

`enrich(step_num: int, project_state: dict, wm: WorkingMemory) -> None`¶

`RuleEvolver`¶

`evolve(current_rules: list[dict], case_bank: CaseBank, k_recent: int = 10) -> list[dict]`¶

`SOARInput(BaseModel)`¶

`SOAROutput(BaseModel)`¶

`CogArchSOARBlock(AIBlock[SOARInput, SOAROutput, dict])`¶

`infer(data: SOARInput) -> Result[SOAROutput]`¶

`SoarSolverBridge`¶

`solve() -> SolverOutput`¶

`get_chunks() -> list[dict]`¶

`GateAction(Enum)`¶

`GateResult`¶

`VerificationResult`¶

`StepVerifier`¶

`introspect_and_gate(step_num: int, project_state: dict) -> GateResult`¶

`verify_and_cross_check(step_num: int, step_output: dict, project_state: dict) -> VerificationResult`¶

`verify_step(step_index: int, step_output: dict, context: dict) -> dict`¶

`build_operator_rules() -> list[dict]`¶

`mirror_to_wm(wm: WorkingMemory, step_num: int, status: str, output: dict) -> None`¶