Trajectory and readers

QitOS public API: trajectory

default_reader selects the canonical journal and supported historical readers. Reading and replay are observation operations, not execution or recovery. Export with REDACTED_PUBLIC explicitly declares loss; reimporting the same count of records is not a raw-data equivalence proof. StoreTrajectoryReader.from_journal and the existing file/default adapters expose read_page(query, cursor=None). page.records, page.next_cursor and page.watermark describe one fixed snapshot. Warm pages hash all snapshot bytes but decode only addressed frames; a temporary disk index and bounded caches avoid retaining the full history. Old complete reads and re-import remain materializing. export_file stages and atomically replaces a canonical JSON file only after final verification; failure preserves existing output. The qit/qita command reference documents exact CLI usage and exit behavior.

完整可运行教程 / 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.

Continuous iter_records and file export validate the full snapshot at the start and end, and each addressed frame during traversal. Iterator values are partial until exhaustion; early close releases resources but does not establish a completed traversal. File output is published only after final validation. Independent read_page calls retain full-history verification on every page.

ExchangeLog

from qitos.core.conversation import ExchangeLog

Source @ f7d4b2d

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

view = PrivacyView.REDACTED_PUBLIC
print(view.value)

Named projections over canonical raw data.

StoreTrajectoryReader

from qitos.tracing.readers import StoreTrajectoryReader

Source @ f7d4b2d

Usage and executable example

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

Usage and executable example

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

Usage and executable example

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

Usage and executable example

The selected source has no bounded snapshot capability.

CursorRejected

from qitos.tracing.paging import CursorRejected

Source @ f7d4b2d

Usage and executable example

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

iter_records

from qitos.tracing.paging import iter_records

Source @ f7d4b2d

Usage and executable example

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

Usage and executable example

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

Usage and executable example

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

Usage and executable example

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

Usage and executable example

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

Usage and executable example

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