Context Engine¶

Context Engine — mvp.context_engine

Cluster: Context & Retrieval | Type: component | MCP Tools: 38

Overview¶

Stateful context tracker and packet builder that maintains an 8-dimensional context vector encoding confidence, observation density, domain stability, constraint pressure, energy level, novelty, change rate, and recency. Updates are persisted to SQLite when a db_path is provided, enabling durable situation awareness across process restarts. The block auto-classifies each context snapshot as stable, unstable, novel, or fatigued using threshold rules on the vector components.

Runtime surfaces:

Core context ops: update/get/classify/detect novelty, constraints, domain transitions, history, reset, and MCP metadata.
MCP persistence ops: context snapshots, constraint sets, search, trend analysis, comparison, stats, import/export, and source_capabilities.
Packet ops: bounded ContextPacket assembly, scorecards, compaction summaries, repository-handle loading, retrieval logs, scorecards, and combined artifact writes.

Reliability/degradation model:

Public packet and MCP outputs expose completion_state, warning_card, evidence, request_id, task_id, and run_id where applicable.
verified means no scorecard hard warnings and no degraded reranker evidence.
qualified-draft covers stale context, budget pruning, reranker fallback/divergence, and capability-pending source handles.
blocked-escalated covers forged handles, path traversal, invalid constraint types, validation failures, and operation failures.

Source capability status:

file_path and grep_result repository handles are operationally populated by the current keyword collector.
symbol, dep_edge, and test_name are declared semantic targets but remain capability_pending until source-index backends are enabled.
The agentic reranker is advisory-only: it can reorder keyword-matched handles, but cannot add handles, bypass budgets, bypass pruning, or bypass load_repository_handle path/candidate guards.

Persistence:

SQLite tables cover context snapshots, constraint sets, context history, novelty log, domain transitions, and errors.
Constraint records are additive: new records carry validated type, priority, authority source, freshness timestamp, active flag, and value; old stored constraint sets load with priority="normal" and authority_source="unknown".

Contract status: qualified-draft until this metadata is reviewed against live product tests beyond the component-level pytest suite.

When to use:

Tracking the current operational context in a physical AI or agent loop
Detecting when a situation has changed significantly enough to require a plan revision
Enforcing runtime constraints and querying which are currently active
Building up a history of context snapshots for downstream novelty detection or audit

Pilot-ready, not standalone regulated assurance

context_engine is suitable for MVP launch, local MCP use, and design-partner pilots where durable context, snapshot restore, constraint tracking, and audit-friendly history are useful. Its situation labels, novelty scores, and change detection are heuristic signals, not calibrated risk or compliance decisions. Before relying on it for regulated, safety-critical, or high-volume production workflows, add domain-specific observation validation, database retention/backup policy, representative evaluation sets, human-review breakpoints, and external legal/compliance review where applicable.

Example:

from mvp.context_engine import ContextSituationBlock, ContextEngineInput

block = ContextSituationBlock(name="ctx", novelty_threshold=0.4)
result = block.infer(ContextEngineInput(
    op="update_context",
    observations=[{"confidence": 0.9, "sensor": "camera"}],
    domain="pick_and_place",
))
# result.ok → True; result.value.situation → "stable" or "novel"

Works well with: affordance_kb, csf, goal_engine

Public API¶

`ContextSituationBlock(AIBlock[ContextEngineInput, ContextEngineOutput, dict])`¶

Context tracking, situation classification, and novelty detection.

Field	Type	Default
`name`	`str`	'context_engine'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	''
`novelty_threshold`	`float`	0.5
`novelty_situation_cutoff`	`float`	0.6
`energy_recovery_rate`	`float`	0.02
`max_history`	`int`	500
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: ContextEngineInput) -> Result[ContextEngineOutput]`¶

`close() -> None`¶

Close the SQLite connection if open. Safe to call multiple times.

`ContextPacketAssembler`¶

Build and maintain bounded, provenance-carrying context packets.

Constructor:

Parameter	Type	Default
`repo_root`	`str \\| Path`	'.'
`context_root`	`str \\| Path`	'.g6/context'

Methods:

`build_context_packet(run_id: str, task: str, phase: str = '', role: str = 'orchestrator', components: list[str] | None = None, repository_query: str = '', budget_tokens: int = 8000, output_contract: str = '', request_id: str = '', task_id: str = '') -> ContextPacket`¶

`retrieve_authority_context(run_id: str) -> list[ContextItem]`¶

`retrieve_component_context(components: list[str], run_id: str) -> list[ContextItem]`¶

`retrieve_repository_context(query: str, budget: int = 2000) -> list[RepositoryHandle]`¶

Return budget-bounded repository handles for query.

`load_file(handle: RepositoryHandle, max_chars: int = 20000) -> str`¶

`is_keyword_candidate(handle: RepositoryHandle, query: str, budget: int = 4000) -> bool`¶

True iff handle is a genuine keyword candidate for query in the LIVE repo.

`retrieve_exemplar_context(run_id: str, query: str) -> list[ContextItem]`¶

`retrieve_memory_context(run_id: str, query: str) -> list[ContextItem]`¶

`retrieve_verification_context(run_id: str) -> list[ContextItem]`¶

`compact_context(packet: ContextPacket, trigger: str, phase: str = '') -> Path`¶

`score_context_packet(packet: ContextPacket) -> ContextScorecard`¶

`envelope_from_scorecard(packet: ContextPacket, scorecard: ContextScorecard) -> dict`¶

Map scorecard and ranker evidence into the public reliability envelope.

`write_retrieval_log(packet: ContextPacket) -> Path`¶

`write_scorecard(packet: ContextPacket) -> Path`¶

`update_memory_from_run(run_id: str, summary: str) -> Path`¶

`subagent_context_packet(packet: ContextPacket, agent_id: str, task: str) -> ContextPacket`¶

`synthesize_subagent_results(results: list[SubagentResult]) -> ContextItem`¶

`ensure_workspace() -> None`¶

`ContextItem(BaseModel)`¶

Single context datum with provenance, freshness, and authority metadata.

Field	Type	Default
`key`	`str`	required
`value`	`str`	required
`section`	`str`	required
`authority`	`AuthorityName`	'model_generated'
`source`	`str`	''
`created_at`	`str`	Field(default_factory=utc_now_iso)
`last_verified_at`	`str`	Field(default_factory=utc_now_iso)
`freshness_class`	`Literal['live', 'fresh', 'aging', 'stale', 'unknown']`	'unknown'
`stale_after`	`str`	''
`retrieved_by`	`str`	''
`retrieved_at`	`str`	Field(default_factory=utc_now_iso)
`included_because`	`str`	''
`outcome`	`str`	''
`token_estimate`	`int`	0
`metadata`	`dict`	Field(default_factory=dict)
`scope`	`str`	''
`validity_interval`	`tuple[str, str] \\| None`	None
`entities`	`tuple[str, ...]`	()

Methods:

`authority_level() -> AuthorityLevel`¶

`is_stale() -> bool`¶

`RepositoryHandle(BaseModel)`¶

Lightweight repository context handle; content is loaded on demand.

Field	Type	Default
`kind`	`Literal['file_path', 'symbol', 'dep_edge', 'test_name', 'grep_result']`	required
`file_path`	`str`	required
`symbol`	`str`	''
`line`	`int \\| None`	None
`snippet`	`str`	''
`token_estimate`	`int`	0
`metadata`	`dict`	Field(default_factory=dict)

`ContextPacket(BaseModel)`¶

Role-specific context bundle sent to an agent invocation.

Field	Type	Default
`run_id`	`str`	required
`phase`	`str`	''
`role`	`str`	'orchestrator'
`authority`	`list[ContextItem]`	Field(default_factory=list)
`task`	`list[ContextItem]`	Field(default_factory=list)
`component`	`list[ContextItem]`	Field(default_factory=list)
`repository`	`list[ContextItem]`	Field(default_factory=list)
`exemplar`	`list[ContextItem]`	Field(default_factory=list)
`memory`	`list[ContextItem]`	Field(default_factory=list)
`tool`	`list[ContextItem]`	Field(default_factory=list)
`verification`	`list[ContextItem]`	Field(default_factory=list)
`output_contract`	`list[ContextItem]`	Field(default_factory=list)
`repository_handles`	`list[RepositoryHandle]`	Field(default_factory=list)
`budget_tokens`	`int`	8000
`used_tokens`	`int`	0
`metadata`	`dict`	Field(default_factory=dict)
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`list[dict]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''

Methods:

`all_items() -> list[ContextItem]`¶

`stale_items() -> list[ContextItem]`¶

`section_items(section: str) -> list[ContextItem]`¶

`ContextScorecard(BaseModel)`¶

Field	Type	Default
`run_id`	`str`	required
`item_count`	`int`	required
`used_tokens`	`int`	required
`budget_tokens`	`int`	required
`budget_ratio`	`float`	required
`stale_count`	`int`	required
`conflict_count`	`int`	required
`missing_sections`	`list[str]`	Field(default_factory=list)
`redundant_count`	`int`	0
`snr_score`	`float`	1.0
`warnings`	`list[str]`	Field(default_factory=list)
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`list[dict]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''

`SubagentResult(BaseModel)`¶

Field	Type	Default
`agent_id`	`str`	required
`task`	`str`	required
`summary`	`str`	required
`evidence`	`list[str]`	Field(default_factory=list)
`unresolved`	`list[str]`	Field(default_factory=list)
`output_label`	`str`	'supported'

`ContextEngineInput(BaseModel)`¶

Input to ContextSituationBlock.

Field	Type	Default
`op`	`Literal['update_context', 'get_context', 'classify_situation', 'detect_change', 'add_constraint', 'remove_constraint', 'list_constraints', 'assess_novelty', 'get_relational_params', 'set_domain', 'get_domain_transitions', 'get_mcp', 'get_history', 'reset', 'ops', 'help']`	'get_context'
`observations`	`list[dict]`	Field(default_factory=list)
`domain`	`str`	''
`constraint_id`	`str`	''
`constraint_type`	`str`	''
`constraint_value`	`str`	''
`constraint_priority`	`str`	''
`authority_source`	`str`	''
`freshness_timestamp`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`context_vector`	`list[float]`	Field(default_factory=list)
`metadata`	`dict`	Field(default_factory=dict)

`ContextEngineOutput(BaseModel)`¶

Output from ContextSituationBlock.

Field	Type	Default
`op`	`str`	required
`context_vector`	`list[float]`	Field(default_factory=list)
`situation`	`str`	''
`confidence`	`float`	0.0
`is_novel`	`bool`	False
`novelty_score`	`float`	0.0
`constraints`	`list[dict]`	Field(default_factory=list)
`change_detected`	`bool`	False
`change_magnitude`	`float`	0.0
`domain`	`str`	''
`history`	`list[dict]`	Field(default_factory=list)
`metadata`	`dict`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`list[dict]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`constraint_priority`	`str`	''
`authority_source`	`str`	''
`freshness_timestamp`	`str`	''

`ContextEngineMCPBlock(AIBlock[MCPContextEngineInput, MCPContextEngineOutput, dict])`¶

36-op context engine MCP block with SQLite persistence.

Field	Type	Default
`name`	`str`	'context_engine_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	field(default_factory=lambda: os.environ.get('CONTEXT_ENGINE_DB_PATH', _DEFAULT_DB))
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPContextEngineInput) -> Result[MCPContextEngineOutput]`¶

`MCPContextEngineInput(BaseModel)`¶

Input to ContextEngineMCPBlock — 36 ops.

Field	Type	Default
`op`	Literal['update_context', 'get_context', 'classify_situation', 'detect_change', 'add_constraint', 'remove_constraint', 'list_constraints', 'assess_novelty', 'get_relational_params', 'set_domain', 'get_history', 'reset', 'get_domain_transitions', 'get_mcp', 'store_context_snapshot', 'load_context_snapshot', 'list_snapshots', 'store_constraint_set', 'load_constraint_set', 'list_constraint_sets', 'search_contexts', 'analyze_context_trend', 'compare_contexts', 'get_context_stats', 'export_context', 'import_context', 'get_info', 'source_capabilities', 'list_patterns', 'build_context_packet', 'score_context_packet', 'compact_context_packet', 'load_repository_handle', 'write_retrieval_log', 'write_scorecard', 'write_context_artifacts', 'ops', 'help']	required
`observations`	`list[dict]`	Field(default_factory=list)
`domain`	`str`	''
`constraint_id`	`str`	''
`constraint_type`	`str`	''
`constraint_value`	`str`	''
`constraint_priority`	`str`	''
`authority_source`	`str`	''
`freshness_timestamp`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`context_vector`	`list[float]`	Field(default_factory=list)
`metadata`	`dict`	Field(default_factory=dict)
`snapshot_name`	`str`	''
`constraint_set_name`	`str`	''
`query`	`str`	''
`top_k`	`int`	10
`context_json`	`str`	''
`compare_id`	`str`	''
`run_id`	`str`	''
`task`	`str`	''
`phase`	`str`	''
`role`	`str`	'orchestrator'
`components`	`list[str]`	Field(default_factory=list)
`repository_query`	`str`	''
`budget_tokens`	`int`	8000
`output_contract`	`str`	''
`packet_json`	`str`	''
`handle_json`	`str`	''
`trigger`	`str`	''
`max_chars`	`int`	20000

`MCPContextEngineOutput(BaseModel)`¶

Output from ContextEngineMCPBlock.

Field	Type	Default
`op`	`str`	required
`context_vector`	`list[float]`	Field(default_factory=list)
`situation`	`str`	''
`confidence`	`float`	0.0
`is_novel`	`bool`	False
`novelty_score`	`float`	0.0
`constraints`	`list[dict]`	Field(default_factory=list)
`change_detected`	`bool`	False
`change_magnitude`	`float`	0.0
`domain`	`str`	''
`history`	`list[dict]`	Field(default_factory=list)
`metadata`	`dict`	Field(default_factory=dict)
`message`	`str`	''
`count`	`int`	0
`retrieved`	`list[dict]`	Field(default_factory=list)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`list[dict]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`source_capabilities`	`dict`	Field(default_factory=dict)
`constraint_priority`	`str`	''
`authority_source`	`str`	''
`freshness_timestamp`	`str`	''
`consistency_radius`	`float \\| None`	None
`consistency_obstruction_count`	`int \\| None`	None
`consistency_gluable`	`bool \\| None`	None
`consistency_edge_residuals`	`dict \\| None`	None
`consistency_coverage_ratio`	`float \\| None`	None
`consistency_corroborated_keys`	`int \\| None`	None
`consistency_uncorroborated_keys`	`int \\| None`	None

`ContextEngineStore`¶

SQLite-backed store for the context_engine MCP sub-package.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`save_snapshot(name: str, vector: list[float], situation: str = '', domain: str = '', metadata: dict | None = None) -> str`¶

`load_snapshot(name: str) -> dict | None`¶

`list_snapshots(limit: int = 50) -> list[dict]`¶

`save_constraint_set(name: str, constraints: list[dict], tags: str = '') -> str`¶

`load_constraint_set(name: str) -> dict | None`¶

`list_constraint_sets(limit: int = 50) -> list[dict]`¶

`log_context(vector: list[float], situation: str = '', domain: str = '', obs_count: int = 0, confidence: float = 0.0, sources: list[dict] | None = None) -> str`¶

`get_history(limit: int = 100) -> list[dict]`¶

`search_contexts(query: str, limit: int = 10) -> list[dict]`¶

Search context history by domain or situation substring.

`log_novelty(vector: list[float], novelty_score: float, is_novel: bool) -> str`¶

`log_domain_transition(old_domain: str, new_domain: str) -> str`¶

`get_domain_transitions(limit: int = 50) -> list[dict]`¶

`log_error(op: str, error_message: str, params: dict | None = None) -> None`¶

`count_all() -> dict[str, int]`¶

Functions¶

`merge_context_items(items: list[ContextItem]) -> list[ContextItem]`¶

Merge by key, retaining the highest-authority item for each key.

MCP Tools¶

Operation	Source
`update_context`	context_engine_mcp
`get_context`	context_engine_mcp
`classify_situation`	context_engine_mcp
`detect_change`	context_engine_mcp
`add_constraint`	context_engine_mcp
`remove_constraint`	context_engine_mcp
`list_constraints`	context_engine_mcp
`assess_novelty`	context_engine_mcp
`get_relational_params`	context_engine_mcp
`set_domain`	context_engine_mcp
`get_history`	context_engine_mcp
`reset`	context_engine_mcp
`get_domain_transitions`	context_engine_mcp
`get_mcp`	context_engine_mcp
`store_context_snapshot`	context_engine_mcp
`load_context_snapshot`	context_engine_mcp
`list_snapshots`	context_engine_mcp
`store_constraint_set`	context_engine_mcp
`load_constraint_set`	context_engine_mcp
`list_constraint_sets`	context_engine_mcp
`search_contexts`	context_engine_mcp
`analyze_context_trend`	context_engine_mcp
`compare_contexts`	context_engine_mcp
`get_context_stats`	context_engine_mcp
`export_context`	context_engine_mcp
`import_context`	context_engine_mcp
`get_info`	context_engine_mcp
`source_capabilities`	context_engine_mcp
`list_patterns`	context_engine_mcp
`build_context_packet`	context_engine_mcp
`score_context_packet`	context_engine_mcp
`compact_context_packet`	context_engine_mcp
`load_repository_handle`	context_engine_mcp
`write_retrieval_log`	context_engine_mcp
`write_scorecard`	context_engine_mcp
`write_context_artifacts`	context_engine_mcp
`ops`	context_engine_mcp
`help`	context_engine_mcp

Context Engine¶

Overview¶

Public API¶

ContextSituationBlock(AIBlock[ContextEngineInput, ContextEngineOutput, dict])¶

infer(data: ContextEngineInput) -> Result[ContextEngineOutput]¶

close() -> None¶

ContextPacketAssembler¶

build_context_packet(run_id: str, task: str, phase: str = '', role: str = 'orchestrator', components: list[str] | None = None, repository_query: str = '', budget_tokens: int = 8000, output_contract: str = '', request_id: str = '', task_id: str = '') -> ContextPacket¶

retrieve_authority_context(run_id: str) -> list[ContextItem]¶

retrieve_component_context(components: list[str], run_id: str) -> list[ContextItem]¶

retrieve_repository_context(query: str, budget: int = 2000) -> list[RepositoryHandle]¶

load_file(handle: RepositoryHandle, max_chars: int = 20000) -> str¶

is_keyword_candidate(handle: RepositoryHandle, query: str, budget: int = 4000) -> bool¶

retrieve_exemplar_context(run_id: str, query: str) -> list[ContextItem]¶

retrieve_memory_context(run_id: str, query: str) -> list[ContextItem]¶

retrieve_verification_context(run_id: str) -> list[ContextItem]¶

compact_context(packet: ContextPacket, trigger: str, phase: str = '') -> Path¶

score_context_packet(packet: ContextPacket) -> ContextScorecard¶

envelope_from_scorecard(packet: ContextPacket, scorecard: ContextScorecard) -> dict¶

write_retrieval_log(packet: ContextPacket) -> Path¶

write_scorecard(packet: ContextPacket) -> Path¶

update_memory_from_run(run_id: str, summary: str) -> Path¶

subagent_context_packet(packet: ContextPacket, agent_id: str, task: str) -> ContextPacket¶

synthesize_subagent_results(results: list[SubagentResult]) -> ContextItem¶

ensure_workspace() -> None¶

ContextItem(BaseModel)¶

authority_level() -> AuthorityLevel¶

is_stale() -> bool¶

RepositoryHandle(BaseModel)¶

ContextPacket(BaseModel)¶

all_items() -> list[ContextItem]¶

stale_items() -> list[ContextItem]¶

section_items(section: str) -> list[ContextItem]¶

ContextScorecard(BaseModel)¶

SubagentResult(BaseModel)¶

ContextEngineInput(BaseModel)¶

ContextEngineOutput(BaseModel)¶

ContextEngineMCPBlock(AIBlock[MCPContextEngineInput, MCPContextEngineOutput, dict])¶

infer(data: MCPContextEngineInput) -> Result[MCPContextEngineOutput]¶

MCPContextEngineInput(BaseModel)¶

MCPContextEngineOutput(BaseModel)¶

ContextEngineStore¶

save_snapshot(name: str, vector: list[float], situation: str = '', domain: str = '', metadata: dict | None = None) -> str¶

load_snapshot(name: str) -> dict | None¶

list_snapshots(limit: int = 50) -> list[dict]¶

save_constraint_set(name: str, constraints: list[dict], tags: str = '') -> str¶

load_constraint_set(name: str) -> dict | None¶

list_constraint_sets(limit: int = 50) -> list[dict]¶

log_context(vector: list[float], situation: str = '', domain: str = '', obs_count: int = 0, confidence: float = 0.0, sources: list[dict] | None = None) -> str¶

get_history(limit: int = 100) -> list[dict]¶

search_contexts(query: str, limit: int = 10) -> list[dict]¶

log_novelty(vector: list[float], novelty_score: float, is_novel: bool) -> str¶

log_domain_transition(old_domain: str, new_domain: str) -> str¶

get_domain_transitions(limit: int = 50) -> list[dict]¶

log_error(op: str, error_message: str, params: dict | None = None) -> None¶

count_all() -> dict[str, int]¶

Functions¶

merge_context_items(items: list[ContextItem]) -> list[ContextItem]¶

MCP Tools¶

`ContextSituationBlock(AIBlock[ContextEngineInput, ContextEngineOutput, dict])`¶

`infer(data: ContextEngineInput) -> Result[ContextEngineOutput]`¶

`close() -> None`¶

`ContextPacketAssembler`¶

`build_context_packet(run_id: str, task: str, phase: str = '', role: str = 'orchestrator', components: list[str] | None = None, repository_query: str = '', budget_tokens: int = 8000, output_contract: str = '', request_id: str = '', task_id: str = '') -> ContextPacket`¶

`retrieve_authority_context(run_id: str) -> list[ContextItem]`¶

`retrieve_component_context(components: list[str], run_id: str) -> list[ContextItem]`¶

`retrieve_repository_context(query: str, budget: int = 2000) -> list[RepositoryHandle]`¶

`load_file(handle: RepositoryHandle, max_chars: int = 20000) -> str`¶

`is_keyword_candidate(handle: RepositoryHandle, query: str, budget: int = 4000) -> bool`¶

`retrieve_exemplar_context(run_id: str, query: str) -> list[ContextItem]`¶

`retrieve_memory_context(run_id: str, query: str) -> list[ContextItem]`¶

`retrieve_verification_context(run_id: str) -> list[ContextItem]`¶

`compact_context(packet: ContextPacket, trigger: str, phase: str = '') -> Path`¶

`score_context_packet(packet: ContextPacket) -> ContextScorecard`¶

`envelope_from_scorecard(packet: ContextPacket, scorecard: ContextScorecard) -> dict`¶

`write_retrieval_log(packet: ContextPacket) -> Path`¶

`write_scorecard(packet: ContextPacket) -> Path`¶

`update_memory_from_run(run_id: str, summary: str) -> Path`¶

`subagent_context_packet(packet: ContextPacket, agent_id: str, task: str) -> ContextPacket`¶

`synthesize_subagent_results(results: list[SubagentResult]) -> ContextItem`¶

`ensure_workspace() -> None`¶

`ContextItem(BaseModel)`¶

`authority_level() -> AuthorityLevel`¶

`is_stale() -> bool`¶

`RepositoryHandle(BaseModel)`¶

`ContextPacket(BaseModel)`¶

`all_items() -> list[ContextItem]`¶

`stale_items() -> list[ContextItem]`¶

`section_items(section: str) -> list[ContextItem]`¶

`ContextScorecard(BaseModel)`¶

`SubagentResult(BaseModel)`¶

`ContextEngineInput(BaseModel)`¶

`ContextEngineOutput(BaseModel)`¶

`ContextEngineMCPBlock(AIBlock[MCPContextEngineInput, MCPContextEngineOutput, dict])`¶

`infer(data: MCPContextEngineInput) -> Result[MCPContextEngineOutput]`¶

`MCPContextEngineInput(BaseModel)`¶

`MCPContextEngineOutput(BaseModel)`¶

`ContextEngineStore`¶

`save_snapshot(name: str, vector: list[float], situation: str = '', domain: str = '', metadata: dict | None = None) -> str`¶

`load_snapshot(name: str) -> dict | None`¶

`list_snapshots(limit: int = 50) -> list[dict]`¶

`save_constraint_set(name: str, constraints: list[dict], tags: str = '') -> str`¶

`load_constraint_set(name: str) -> dict | None`¶

`list_constraint_sets(limit: int = 50) -> list[dict]`¶

`log_context(vector: list[float], situation: str = '', domain: str = '', obs_count: int = 0, confidence: float = 0.0, sources: list[dict] | None = None) -> str`¶

`get_history(limit: int = 100) -> list[dict]`¶

`search_contexts(query: str, limit: int = 10) -> list[dict]`¶

`log_novelty(vector: list[float], novelty_score: float, is_novel: bool) -> str`¶

`log_domain_transition(old_domain: str, new_domain: str) -> str`¶

`get_domain_transitions(limit: int = 50) -> list[dict]`¶

`log_error(op: str, error_message: str, params: dict | None = None) -> None`¶

`count_all() -> dict[str, int]`¶

`merge_context_items(items: list[ContextItem]) -> list[ContextItem]`¶