Context 与 memory 契约

通过显式扩展名注册 contributor、selector、budget policy 与 compactor。context 选择不授予工具权限。CompactionReceipt 声明输入、输出身份和损失，省略不是无损摘要。memory 输入不是 checkpoint store。缺失工厂在组合时失败；可执行教程展示方法参数及实际有损 compaction。完整可运行教程 / Complete tutorial · API index 以下签名和字段由固定源码提取；签名是参考，不是可直接运行的程序。每个条目的源码链接绑定同一 runtime baseline。类型中的 Any 不代表任意对象均受支持，应结合上述行为契约和教程使用。具体 module API 为 qitos.kit.memory.adapter.MemorySourceAdapter(memory, *, namespace, query=None, required=False, priority=0) 和 qitos.kit.context.compaction.ClosedExchangeWindowCompactor()。通过现有 extensions mapping 绑定，不增加 root export 或 Engine 构造参数。Memdir 恢复时明确报告缺失资源（create=True 用于初始化）；Markdown 仅在当前实例内召回。required contribution 的优先级高于 optional contribution。 RequestView 的预算驱动 exchange 省略现在要求显式 compaction policy。compact callback 收到符合省略条件的 IDs 和已选择 exchange projection 的 SHA-256 digest，必须返回对应的损失 receipt。已有 receipt 本身不授权新的省略。request builder 负责最近窗口和全部保护检查，continuation 采用保守保护。Python-only context 预算/损失设置和迁移边界见教程。

StaticContextContributor

from qitos.core.context import StaticContextContributor

Source @ d000a42 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

# 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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

# 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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

# 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 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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

LifecyclePolicy

from qitos.engine.runtime import LifecyclePolicy

Source @ d000a42 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

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