Context and memory contracts

Register contributors, selectors, budget policies and compactors with explicit extension names. Context selection does not grant tool authority. CompactionReceipt declares the input, output identity and losses; omission is not a lossless summary. Memory input is not a checkpoint store. Missing factories fail at composition. See the executable tutorial for method arguments and an observed lossy compaction. 完整可运行教程 / Complete tutorial · API index Signatures and fields below are extracted from the pinned source. Signatures are reference material, not standalone programs. Source links bind the same runtime baseline. Any does not imply arbitrary objects are supported; use the behavioral contract above and the linked tutorial. The concrete module APIs are qitos.kit.memory.adapter.MemorySourceAdapter(memory, *, namespace, query=None, required=False, priority=0) and qitos.kit.context.compaction.ClosedExchangeWindowCompactor(). Bind them using the existing extensions mapping; no root export or Engine constructor option is added. Memdir restore is explicit about missing resources (create=True initializes); Markdown recall is instance-scoped. Required contributions take precedence over optional priorities. RequestView now requires an explicit compaction policy for budget-driven exchange omission. Its compact callback receives eligible omitted IDs and the SHA-256 digest of the selected exchange projection, and must return a matching loss receipt. Preexisting receipts alone do not authorize new omission. The request builder owns the recent window and all protection checks; continuation protection is conservative. See the tutorial for the Python-only context budget/loss settings and migration boundaries.

StaticContextContributor

from qitos.core.context import StaticContextContributor

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

contributor = StaticContextContributor("notes.project", "project", "Use supplied notes only.")

Small reusable contributor for project/user/session context.

Field	Type	Default
`contributor_id`	`str`	`required`
`source`	`str`	`required`
`value`	`Any`	`field(repr=False)`
`priority`	`int`	`0`
`requested_placement`	`str`	`'developer'`
`required`	`bool`	`False`
`persistence_horizon`	`str`	`'request'`
`sensitivity`	`str`	`'internal'`

PriorityContextSelectionPolicy

from qitos.core.context import PriorityContextSelectionPolicy

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

class AuditedSelector(PriorityContextSelectionPolicy):
    def select(self, contributions, **options):
        contributions = tuple(contributions)
        print([item.contribution_id for item in contributions])
        return super().select(contributions, **options)

Default deterministic priority/identity selection policy.

Field	Type	Default
`policy_id`	`str`	`'qitos.context.priority/v1'`

PriorityContextSelectionPolicy.select

select(contributions: Iterable[ContextContribution], *, budget: ContextBudget, already_used_units: int=0, counter: UnitCounter=default_unit_counter) -> ContextSelection

Parameter	Type	Default
`contributions`	`Iterable[ContextContribution]`	`required`
`budget`	`ContextBudget`	`required`
`already_used_units`	`int`	`0`
`counter`	`UnitCounter`	`default_unit_counter`

Source

DeclaredContextBudgetPolicy

from qitos.core.context import DeclaredContextBudgetPolicy

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

policy = DeclaredContextBudgetPolicy(default_max_input_units=4096)

Use adapter-declared capacity; never infer it from a model name.

Field	Type	Default
`default_max_input_units`	`int`	`120000`
`unit`	`str`	`'characters'`
`protected_recent_exchanges`	`int`	`1`
`policy_id`	`str`	`'qitos.context.declared_budget/v1'`

CompactionReceipt

from qitos.core.request_view import CompactionReceipt

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

# receipt is returned by the compactor in the context tutorial.
print(receipt.declared_losses, receipt.output_digest)

Field	Type	Default
`receipt_id`	`str`	`required`
`input_exchange_ids`	`tuple[str, ...]`	`required`
`output_digest`	`str`	`required`
`policy_id`	`str`	`required`
`declared_losses`	`tuple[str, ...]`	`()`
`model_reference`	`Optional[str]`	`None`

MemoryRecord

from qitos.core.memory import MemoryRecord

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

record = MemoryRecord("user", "remembered-value=17", 0)

One logical memory fact; independent equal facts receive distinct IDs.

Field	Type	Default
`role`	`str`	`required`
`content`	`Any`	`required`
`step_id`	`int`	`required`
`metadata`	`Dict[str, Any]`	`field(default_factory=dict)`
`record_id`	`str`	`field(default_factory=lambda: uuid4().hex)`

MemoryResourceError

from qitos.core.memory import MemoryResourceError

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

# Restore raises MemoryResourceError if the bound root is missing.

A bound durable memory resource is missing or unavailable.

MemdirMemory

from qitos.kit.memory.memdir_memory import MemdirMemory

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

memory = MemdirMemory("./project-memory", create=True)
memory.append(MemoryRecord("user", "remembered-value=17", 0))
# Later process: MemdirMemory("./project-memory")

Persist text records with stable identity and fresh disk retrieval.

Restore is the default and fails if a bound root is missing. Pass
``create=True`` only to explicitly initialize a namespace. ``reset`` and
``evict`` affect the append-time cache only; ``delete`` removes a local fact.
Arbitrary Python/JSON content and metadata are not a durable round-trip API.

MemdirMemory(memory_dir: str='.qitos/memory', *, global_memory_dir: str | None=None, create: bool=False, max_index_entries: int=200, max_index_chars: int=25000) -> Any (see behavior contract)

Parameter	Type	Default
`memory_dir`	`str`	`'.qitos/memory'`
`global_memory_dir`	`str \| None`	`None`
`create`	`bool`	`False`
`max_index_entries`	`int`	`200`
`max_index_chars`	`int`	`25000`

MemdirMemory.append

append(record: MemoryRecord) -> None

Parameter	Type	Default
`record`	`MemoryRecord`	`required`

Source

MemdirMemory.retrieve

retrieve(query: Optional[Dict[str, Any]]=None, state: Any=None, observation: Any=None) -> List[MemoryRecord]

Parameter	Type	Default
`query`	`Optional[Dict[str, Any]]`	`None`
`state`	`Any`	`None`
`observation`	`Any`	`None`

Source

MemdirMemory.summarize

summarize(max_items: int=30) -> str

Parameter	Type	Default
`max_items`	`int`	`30`

Render current records, never stale index snippets after external edits.

Source

MemdirMemory.reset

reset(run_id: Optional[str]=None) -> None

Parameter	Type	Default
`run_id`	`Optional[str]`	`None`

Source

MemdirMemory.evict

evict() -> int

Source

MemdirMemory.delete

delete(record_id: str) -> bool

Parameter	Type	Default
`record_id`	`str`	`required`

Forget one logical record in this namespace, never global memory.

Idempotent for an absent identity. This is local-file deletion, not
secure erasure or a concurrent multi-file transaction. Callers serialize
writes to one Memdir namespace; use a transactional backend otherwise.

Source

MarkdownFileMemory

from qitos.kit.memory.markdown_file_memory import MarkdownFileMemory

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

memory = MarkdownFileMemory("./run-memory.md")
# Existing logs are not reloaded; recall is instance-scoped.

Persist memory records in a local markdown log file.

MarkdownFileMemory(path: str='memory.md', max_in_memory: int=200) -> Any (see behavior contract)

Parameter	Type	Default
`path`	`str`	`'memory.md'`
`max_in_memory`	`int`	`200`

MemorySourceAdapter

from qitos.kit.memory.adapter import MemorySourceAdapter

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

source = MemorySourceAdapter(memory, namespace="project", required=True)
# Bind source as extensions["project_memory"].

Recall a factory-bound namespace as user-level context data.

Namespace is a logical label, not a directory selector or an authorization.
The factory must bind the correct Memory resource. No global memory is read
by this adapter. Every request recalls fresh data, including after restore;
revision is provenance, never a permanent suppression cache.

MemorySourceAdapter(memory: Memory, *, namespace: str, query: Mapping[str, Any] | None=None, required: bool=False, priority: int=0) -> None

Parameter	Type	Default
`memory`	`Memory`	`required`
`namespace`	`str`	`required`
`query`	`Mapping[str, Any] \| None`	`None`
`required`	`bool`	`False`
`priority`	`int`	`0`

MemorySourceAdapter.contribute

contribute(request: Any) -> Sequence[ContextContribution]

Parameter	Type	Default
`request`	`Any`	`required`

Source

MemorySourceAdapter.reset

reset(run_id: str | None=None) -> None

Parameter	Type	Default
`run_id`	`str \| None`	`None`

No owned state; never reset the borrowed memory.

Source

MemorySourceAdapter.close

close() -> None

No owned resources; never close the borrowed memory.

Source

ClosedExchangeWindowCompactor

from qitos.kit.context.compaction import ClosedExchangeWindowCompactor

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

extensions["closed_window"] = ClosedExchangeWindowCompactor
# Configure compaction={"provider": "closed_window"}; the policy supplies receipts.

Omit the oldest eligible closed exchanges without a summary.

RequestView owns eligibility and the ContextBudget recent-window boundary.
This stateless policy authorizes that derived selection and records its loss;
it never modifies an ExchangeLog or widens a budget or codec loss setting.

ClosedExchangeWindowCompactor.compact

compact(*, exchange_ids: Sequence[str], selected_digest: str, required_units: int, available_units: int) -> CompactionReceipt | None

Parameter	Type	Default
`exchange_ids`	`Sequence[str]`	`required`
`selected_digest`	`str`	`required`
`required_units`	`int`	`required`
`available_units`	`int`	`required`

Source

RequestView

from qitos.core.request_view import RequestView

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

# Advanced direct projection; normal users select the configured compactor.
view = RequestView.from_exchange_log(log, compaction_policy=ClosedExchangeWindowCompactor())

One immutable, deterministic provider-neutral model request view.

Field	Type	Default
`request_id`	`str`	`required`
`target`	`RequestTarget`	`required`
`source_log_id`	`str`	`required`
`source_log_digest`	`str`	`required`
`selected_items_json`	`tuple[str, ...]`	`field(repr=False)`
`instructions_json`	`tuple[str, ...]`	`field(default=(), repr=False)`
`tool_schemas_json`	`tuple[str, ...]`	`field(default=(), repr=False)`
`context_json`	`tuple[str, ...]`	`field(default=(), repr=False)`
`capability_requirements`	`tuple[str, ...]`	`()`
`reasoning_policy`	`ReasoningPolicy`	`field(default_factory=ReasoningPolicy)`
`continuation`	`Optional[ContinuationRef]`	`None`
`context_budget`	`ContextBudget`	`field(default_factory=ContextBudget)`
`selection`	`SelectionReport`	`field(default_factory=lambda: SelectionReport((), (), (), (), (), (), (), (), 0, 0, 0, 'characters'))`
`compaction_receipts`	`tuple[CompactionReceipt, ...]`	`()`
`artifact_refs`	`tuple[ArtifactRef, ...]`	`()`
`steering_boundary_json`	`str`	`'{}'`
`correlation_facts_json`	`tuple[str, ...]`	`field(default=(), repr=False)`
`provenance_json`	`str`	`'{}'`
`protocol_id`	`str`	`'unknown'`
`tool_use_policy`	`str`	`'auto'`
`tool_use_satisfied`	`bool`	`False`
`schema_version`	`str`	`REQUEST_VIEW_SCHEMA_VERSION`

RequestView.from_exchange_log

from_exchange_log(log: ExchangeLog, *, model: Any=None, target: Optional[RequestTarget]=None, instructions: Iterable[Mapping[str, Any]]=(), tool_schemas: Iterable[Mapping[str, Any]]=(), capability_requirements: Iterable[str]=(), reasoning_policy: Optional[ReasoningPolicy]=None, continuation: Optional[ContinuationRef]=None, context_budget: Optional[ContextBudget]=None, context_contributions: Iterable[ContextContribution]=(), context_selection_policy: Any=None, context_unit_counter: Any=None, compaction_receipts: Iterable[CompactionReceipt]=(), compaction_policy: Any=None, artifact_refs: Iterable[ArtifactRef]=(), available_artifact_ids: Optional[Iterable[str]]=None, protocol_id: str='unknown', tool_use_policy: str='auto', tool_use_satisfied: bool=False) -> 'RequestView'

Parameter	Type	Default
`log`	`ExchangeLog`	`required`
`model`	`Any`	`None`
`target`	`Optional[RequestTarget]`	`None`
`instructions`	`Iterable[Mapping[str, Any]]`	`()`
`tool_schemas`	`Iterable[Mapping[str, Any]]`	`()`
`capability_requirements`	`Iterable[str]`	`()`
`reasoning_policy`	`Optional[ReasoningPolicy]`	`None`
`continuation`	`Optional[ContinuationRef]`	`None`
`context_budget`	`Optional[ContextBudget]`	`None`
`context_contributions`	`Iterable[ContextContribution]`	`()`
`context_selection_policy`	`Any`	`None`
`context_unit_counter`	`Any`	`None`
`compaction_receipts`	`Iterable[CompactionReceipt]`	`()`
`compaction_policy`	`Any`	`None`
`artifact_refs`	`Iterable[ArtifactRef]`	`()`
`available_artifact_ids`	`Optional[Iterable[str]]`	`None`
`protocol_id`	`str`	`'unknown'`
`tool_use_policy`	`str`	`'auto'`
`tool_use_satisfied`	`bool`	`False`

Source

ContextBudgetExceededError

from qitos.core.request_view import ContextBudgetExceededError

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

# Required/protected exchanges cannot be removed to make a request fit.

LifecyclePolicy

from qitos.engine.runtime import LifecyclePolicy

Source @ d000a42 Usage and executable example Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

class PauseFirstTool(LifecyclePolicy):
    policy_id = "notes.pause"
    def should_pause(self, context):
        return context.step_id == 0

Replaceable cooperative lifecycle policy; never a worker scheduler.

LifecyclePolicy.should_pause

should_pause(context: RuntimeSnapshotContext) -> bool

Parameter	Type	Default
`context`	`RuntimeSnapshotContext`	`required`

Source

​StaticContextContributor

​PriorityContextSelectionPolicy

​PriorityContextSelectionPolicy.select

​DeclaredContextBudgetPolicy

​CompactionReceipt

​MemoryRecord

​MemoryResourceError

​MemdirMemory

​MemdirMemory.append

​MemdirMemory.retrieve

​MemdirMemory.summarize

​MemdirMemory.reset

​MemdirMemory.evict

​MemdirMemory.delete

​MarkdownFileMemory

​MemorySourceAdapter

​MemorySourceAdapter.contribute

​MemorySourceAdapter.reset

​MemorySourceAdapter.close

​ClosedExchangeWindowCompactor

​ClosedExchangeWindowCompactor.compact

​RequestView

​RequestView.from_exchange_log

​ContextBudgetExceededError

​LifecyclePolicy

​LifecyclePolicy.should_pause

StaticContextContributor

PriorityContextSelectionPolicy

PriorityContextSelectionPolicy.select

DeclaredContextBudgetPolicy

CompactionReceipt

MemoryRecord

MemoryResourceError

MemdirMemory

MemdirMemory.append

MemdirMemory.retrieve

MemdirMemory.summarize

MemdirMemory.reset

MemdirMemory.evict

MemdirMemory.delete

MarkdownFileMemory

MemorySourceAdapter

MemorySourceAdapter.contribute

MemorySourceAdapter.reset

MemorySourceAdapter.close

ClosedExchangeWindowCompactor

ClosedExchangeWindowCompactor.compact

RequestView

RequestView.from_exchange_log

ContextBudgetExceededError

LifecyclePolicy

LifecyclePolicy.should_pause