Trajectory 与 reader

QitOS public API: trajectory

default_reader 选择 canonical journal 和支持的历史 reader。读取与回放是观察，不执行或恢复 Agent。REDACTED_PUBLIC 导出明确声明损失；重新导入后记录数相同不能证明 raw 数据等价。旧完整读取仍会全量加载 journal。qit/qita 命令参考说明 CLI 的实际用法与退出行为。

完整可运行教程 / Complete tutorial · API index

以下签名和字段由固定源码提取；签名是参考，不是可直接运行的程序。每个条目的源码链接绑定同一 runtime baseline。类型中的 Any 不代表任意对象均受支持，应结合上述行为契约和教程使用。

StoreTrajectoryReader.from_journal 与现有 file/default adapter 提供 read_page(query, cursor=None)。page.records、page.next_cursor、page.watermark 描述固定快照。warm page 仍 hash 全部快照字节，但仅解析寻址 frame；临时磁盘索引和固定 cache 避免保留全历史。旧完整读取和 re-import 仍物化。export_file 在最终校验后原子替换文件，失败保留原目标。

连续 iter_records 与文件导出在开始、结束时校验完整快照，遍历时逐 frame 校验。迭代值在耗尽前属于 partial；提前 close 释放资源，但不证明遍历完成。文件仅在最终校验后发布。独立 read_page 每页仍校验全部历史字节。

​

ExchangeLog

from qitos.core.conversation import ExchangeLog

Source @ f7d4b2d

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

log = ExchangeLog.from_dict(conversation_payload)

Append-only persistent exchange facts and queued safe-boundary steering.

ExchangeLog(log_id: Optional[str]=None, *, items: Optional[Iterable[ExchangeItem]]=None, queued_steering: Optional[Iterable[SteeringItem]]=None, schema_version: str=EXCHANGE_LOG_SCHEMA_VERSION) -> None

Parameter	Type	Default
log_id	Optional[str]	None
items	Optional[Iterable[ExchangeItem]]	None
queued_steering	Optional[Iterable[SteeringItem]]	None
schema_version	str	EXCHANGE_LOG_SCHEMA_VERSION

​

ExchangeLog.from_dict

from_dict(payload: Mapping[str, Any]) -> 'ExchangeLog'

Parameter	Type	Default
payload	Mapping[str, Any]	required

Source

​

ExchangeLog.declared_calls

declared_calls(batch_id: str) -> List[ToolCall]

Parameter	Type	Default
batch_id	str	required

Source

​

ExchangeLog.results_for_batch

results_for_batch(batch_id: str) -> List[ToolResultItem]

Parameter	Type	Default
batch_id	str	required

Source

​

ExchangeLog.results_for_batch_in_declaration_order

results_for_batch_in_declaration_order(batch_id: str) -> List[ToolResultItem]

Parameter	Type	Default
batch_id	str	required

Derive results in call declaration order without rewriting facts.

Source

​

ExchangeLog.open_batch_id

open_batch_id() -> Optional[str]

Source

​

JournalTrajectoryStore

from qitos.tracing.journal_store import JournalTrajectoryStore

Source @ f7d4b2d

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

store = JournalTrajectoryStore("trajectory.journal", read_only=True)

Durable local store with framed append, recovery, and a derived index.

JournalTrajectoryStore(path: str | Path, *, recover_partial_tail: bool=True, max_query_records: int=10000, read_only: bool=False) -> None

Parameter	Type	Default
path	str | Path	required
recover_partial_tail	bool	True
max_query_records	int	10000
read_only	bool	False

​

JournalTrajectoryStore.read_session

read_session(session_id: str) -> Trajectory

Parameter	Type	Default
session_id	str	required

Source

​

default_reader

from qitos.qita.reader import default_reader

Source @ f7d4b2d

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

reader = default_reader(root)
trajectory = reader.read_session(identity, view=PrivacyView.RAW_PRIVATE)
print(len(trajectory.records))

Select canonical data with bounded trace compatibility, or explicit rollback.

default_reader(root: str | Path, *, selector: str='trajectory') -> Any

Parameter	Type	Default
root	str | Path	required
selector	str	'trajectory'

​

CanonicalTrajectoryExporter

from qitos.tracing.exporter import CanonicalTrajectoryExporter

Source @ f7d4b2d

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

exporter = CanonicalTrajectoryExporter()
receipt = exporter.export_file(reader, query, root / "public-trajectory.json")
print(receipt.record_count)

Canonical JSON exporter; exact for the selected projection.

​

CanonicalTrajectoryExporter.export

export(trajectory: Trajectory, *, view: PrivacyView=PrivacyView.REDACTED_PUBLIC) -> ExportArtifact

Parameter	Type	Default
trajectory	Trajectory	required
view	PrivacyView	PrivacyView.REDACTED_PUBLIC

Source

​

CanonicalTrajectoryExporter.export_file

export_file(reader: Any, query: Any, target: Any, *, view: PrivacyView=PrivacyView.REDACTED_PUBLIC, cancelled: Any=None) -> Any

Parameter	Type	Default
reader	Any	required
query	Any	required
target	Any	required
view	PrivacyView	PrivacyView.REDACTED_PUBLIC
cancelled	Any	None

Export a bounded snapshot through owned staging and atomic replacement.

Source

​

CanonicalTrajectoryExporter.reimport

reimport(artifact: ExportArtifact) -> Trajectory

Parameter	Type	Default
artifact	ExportArtifact	required

Source

​

PrivacyView

from qitos.tracing.trajectory import PrivacyView

Source @ f7d4b2d

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

view = PrivacyView.REDACTED_PUBLIC
print(view.value)

Named projections over canonical raw data.

​

StoreTrajectoryReader

from qitos.tracing.readers import StoreTrajectoryReader

Source @ f7d4b2d

用法与可执行示例

Reader adapter for any conforming trajectory store.

StoreTrajectoryReader(store: TrajectoryStore) -> None

Parameter	Type	Default
store	TrajectoryStore	required

​

StoreTrajectoryReader.from_journal

from_journal(path: str | Path) -> 'StoreTrajectoryReader'

Parameter	Type	Default
path	str | Path	required

Open a validated read-only journal without materializing its history.

Source

​

StoreTrajectoryReader.read_page

read_page(query: TrajectoryQuery, cursor: TrajectoryCursor | None=None, *, view: PrivacyView=PrivacyView.REDACTED_PUBLIC) -> TrajectoryPage

Parameter	Type	Default
query	TrajectoryQuery	required
cursor	TrajectoryCursor | None	None
view	PrivacyView	PrivacyView.REDACTED_PUBLIC

Source

​

StoreTrajectoryReader.close

close() -> None

Source

​

TrajectoryCursor

from qitos.tracing.paging import TrajectoryCursor

Source @ f7d4b2d

用法与可执行示例

Reader-local opaque token. No host path or credential is serialized.

Field	Type	Default
token	str	required

​

TrajectoryPage

from qitos.tracing.paging import TrajectoryPage

Source @ f7d4b2d

用法与可执行示例

Field	Type	Default
records	tuple[TrajectoryRecord, ...]	required
next_cursor	TrajectoryCursor | None	required
snapshot	TrajectoryCursor	required
watermark	int	required

​

BoundedReadUnsupported

from qitos.tracing.paging import BoundedReadUnsupported

Source @ f7d4b2d

用法与可执行示例

The selected source has no bounded snapshot capability.

​

CursorRejected

from qitos.tracing.paging import CursorRejected

Source @ f7d4b2d

用法与可执行示例

A cursor or its source no longer describes the captured snapshot.

​

iter_records

from qitos.tracing.paging import iter_records

Source @ f7d4b2d

用法与可执行示例

Iterate a snapshot; output is partial until exhaustion/final validation.

Early close releases resources but does not claim a completed traversal.
Independent read_page calls always reverify the complete snapshot.

iter_records(reader: Any, query: TrajectoryQuery, *, view: PrivacyView=PrivacyView.REDACTED_PUBLIC) -> Iterator[TrajectoryRecord]

Parameter	Type	Default
reader	Any	required
query	TrajectoryQuery	required
view	PrivacyView	PrivacyView.REDACTED_PUBLIC

​

read_page

from qitos.tracing.paging import read_page

Source @ f7d4b2d

用法与可执行示例

Require bounded capability; never materialize a third-party fallback.

read_page(reader: Any, query: TrajectoryQuery, cursor: TrajectoryCursor | None=None, *, view: PrivacyView=PrivacyView.REDACTED_PUBLIC) -> TrajectoryPage

Parameter	Type	Default
reader	Any	required
query	TrajectoryQuery	required
cursor	TrajectoryCursor | None	None
view	PrivacyView	PrivacyView.REDACTED_PUBLIC

​

TrajectoryQuery

from qitos.tracing.trajectory import TrajectoryQuery

Source @ f7d4b2d

用法与可执行示例

Store-independent declarative trajectory query.

Field	Type	Default
session_id	Optional[str]	None
run_id	Optional[str]	None
work_item_id	Optional[str]	None
kinds	Tuple[RecordKind, ...]	()
after_sequence	Optional[int]	None
limit	Optional[int]	None

​

LossReport

from qitos.tracing.trajectory import LossReport

Source @ f7d4b2d

用法与可执行示例

Explicit fidelity report carried by readers, exporters and evaluators.

Field	Type	Default
policy_id	str	'qitos.loss/none'
entries	Tuple[LossEntry, ...]	()

​

LossReport.from_dict

from_dict(value: Mapping[str, Any]) -> 'LossReport'

Parameter	Type	Default
value	Mapping[str, Any]	required

Source

​

ExportArtifact

from qitos.tracing.exporter import ExportArtifact

Source @ f7d4b2d

用法与可执行示例

Field	Type	Default
exporter_id	str	required
format_version	str	required
content_type	str	required
data	bytes	required
digest	str	required
privacy_view	PrivacyView	required
exact_reimport	bool	required
provenance	Dict[str, Any]	required
loss	LossReport	required

​

FileExportReceipt

from qitos.tracing.streaming import FileExportReceipt

Source @ f7d4b2d

用法与可执行示例

Returned only after full validation, fsync and atomic replacement.

Field	Type	Default
record_count	int	required
size_bytes	int	required
digest	str	required
privacy_view	PrivacyView	required
completed	bool	True