diff --git a/AGENTS.md b/AGENTS.md index a1efeed..492f57d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -37,6 +37,10 @@ The **only** exception is documentation explicitly designated as Chinese: Everything outside that list — including code under `src/`, this file, and all other docs — is English. +Do not label a document as `hand-written` unless its contents are actually +authored and maintained by humans. For agent-authored source-of-truth documents, +use wording such as `Chinese source` instead. + ### Bilingual research notes Research notes are bilingual: diff --git a/README.md b/README.md index 0b6515f..0d55cce 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,12 @@ checks the result to judge. A non-zero exit means the run could not happen at all: `1` for missing credentials or an API that kept refusing, `2` for a misused command line. +Every Agent Run also writes a replayable internal Event Journal under +`~/.nanoPyCodeAgent/journals/`. These JSONL files can contain prompts, model +replies, repository content, and tool results, so treat them as sensitive; +the directory is user-only (`0700`) and each file is `0600`. Journals are not +public run output or trajectories, and they are not rotated automatically yet. + #### Run a branch or tagged version Run an unreleased branch or a specific release tag straight from GitHub: diff --git a/README.zh-CN.md b/README.zh-CN.md index f263cac..bce80d9 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -49,6 +49,11 @@ printf "%s" "$TASK" | nanoPyCodeAgent 完,那该由检查结果的一方去判定。非零退出码表示这次运行根本没能进行:`1` 是缺少 凭据或 API 持续失败,`2` 是命令行用错了。 +每次 Agent Run 还会在 `~/.nanoPyCodeAgent/journals/` 下写入可重放的内部 Event +Journal。这些 JSONL 文件可能包含提示词、模型回复、仓库内容和工具结果,应按敏感 +数据处理;目录只允许当前用户访问(`0700`),每个文件的权限为 `0600`。Journal 既 +不是公开 run output,也不是 trajectory,目前还不会自动轮转。 + #### 运行某个分支或标签版本 直接从 GitHub 运行未发布的分支,或某个具体的发布标签: diff --git a/docs/changelogs/0.8.x.md b/docs/changelogs/0.8.x.md index 06ae71c..6561294 100644 --- a/docs/changelogs/0.8.x.md +++ b/docs/changelogs/0.8.x.md @@ -26,6 +26,12 @@ All notable changes in the **0.8.x** release series are documented here. headless CLI through stdin, forwards model credentials and endpoint configuration, saves the run log under `/logs/agent/`, and reports the installed agent version without adding Harbor to the end-user package. +- Versioned Native Events and a per-run internal Event Journal. Model, tool, + user, and terminal facts are appended as replayable JSONL with stable IDs, + producer versions, ordering, UTC timestamps, durations, usage, provider + response metadata, and explicit truncation metadata. Journals are stored as + sensitive user-only files while the existing text stdout remains unchanged + as an event projection. ### Fixed - Declare `httpx` as a direct runtime dependency so clean and containerized diff --git a/docs/dev_docs/README.md b/docs/dev_docs/README.md new file mode 100644 index 0000000..24a0bd1 --- /dev/null +++ b/docs/dev_docs/README.md @@ -0,0 +1,32 @@ +# System Design Documentation + +This directory contains the implemented architecture, internal protocols, and +other system-design contracts of nanoPyCodeAgent. These documents describe the +system as it exists. Proposals that are still under discussion should be +clearly marked as RFCs; durable decision rationale belongs in an ADR when the +choice is costly to reverse or otherwise surprising. + +System-design documents are bilingual and split by language: + +- [`zh-CN/`](zh-CN/) — **Chinese source** (source of truth) +- [`en/`](en/) — **English, generated from the Chinese source** (do not edit by + hand) + +Write or revise the Chinese source first. Before a pull request containing the +change is opened or updated for review, translate or refresh the entire +corresponding English file. The agent preparing or landing the pull request +must report whether the two versions are in sync. + +These documents use the domain language defined in [`CONTEXT.md`](../../CONTEXT.md). +They complement, rather than replace: + +- [`../research/`](../research/) for pre-implementation investigation; +- [`../dev_notes/`](../dev_notes/) for release-series development notes; and +- [`../superpowers/specs/`](../superpowers/specs/) for implementation plans and + design proposals associated with a particular piece of work. + +## Documents + +- Event Journal Protocol v1: + [English](en/event-journal-protocol-v1.md) | + [Chinese](zh-CN/event-journal-protocol-v1.md) diff --git a/docs/dev_docs/en/event-journal-protocol-v1.md b/docs/dev_docs/en/event-journal-protocol-v1.md new file mode 100644 index 0000000..ce1aef0 --- /dev/null +++ b/docs/dev_docs/en/event-journal-protocol-v1.md @@ -0,0 +1,538 @@ +# Event Journal Implementation Protocol v1 + +> Generated from the Chinese source +> [`../zh-CN/event-journal-protocol-v1.md`](../zh-CN/event-journal-protocol-v1.md). +> Do not edit by hand. + +| Item | Value | +|---|---| +| Status | Implemented | +| Protocol version | v1 | +| `schema_version` | `1` | +| Visibility | Internal protocol; not a public Run Output or ATIF interface | +| Domain terminology | [`CONTEXT.md`](../../../CONTEXT.md) | +| Core implementation | [`event_journal.py`](../../../src/nanopycodeagent/event_journal.py) | +| Event production and text projection | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| Behavioral tests | [`test_event_journal.py`](../../../tests/test_event_journal.py), [`test_agent_events.py`](../../../tests/test_agent_events.py), [`test_agent.py`](../../../tests/test_agent.py) | + +## Document classification + +This is an **implemented protocol specification**, not an RFC. + +- An RFC is for a proposal, discussion, and review before implementation or a + behavioral change. +- An ADR records the decision and rationale for an architectural choice that is + costly to reverse, surprising, or carries significant tradeoffs. +- This document fixes the wire contract, event semantics, persistence behavior, + and compatibility boundary that are already implemented. + +A future behavior change may begin with an RFC or ADR. Once accepted and +implemented, that change must also update this document and the protocol version. + +The terms MUST, SHOULD, and MAY denote protocol requirements, recommendations, +and permitted behavior. Unless stated otherwise, fields and validation rules +describe `schema_version = 1`. + +## Goals and boundaries + +The Event Journal preserves replayable internal runtime facts for one **Agent +Run**. The complete path is: + +```text +Agent core + ↓ produces +Native Event {type, payload} + ├─→ live projector → existing stdout text Run Output + └─→ journal writer → Journal Entry + ↓ append-only UTF-8 JSONL + Event Journal + ↓ future projector (outside v1) + ATIF Trajectory / other public representations +``` + +The boundaries are: + +- A **Native Event** is an agent-independent runtime fact. +- A **Journal Entry** adds the identity, ordering, and recording time required + to persist a Native Event. +- An **Event Journal** is the append sequence of Journal Entries for one Agent + Run. +- stdout is a live text projection of the same Native Events, but it is not the + Event Journal. +- ATIF, `stream-json`, and other public output projections are outside v1. + +The Event Emitter MUST append an event to the Journal before sending it to a +live projector. Persistence truncation applies only to the Journal Entry. The +projector receives the untruncated Native Event, so Journal size limits do not +change existing stdout behavior. + +## Encoding and basic types + +- A file is UTF-8 JSONL. Every complete line contains exactly one Journal Entry + and ends with `\n`. +- The writer uses compact JSON. Object field order has no semantics. +- A payload MUST be a JSON object. Its recursive values are limited to `null`, + booleans, finite numbers, strings, arrays, and objects. +- Non-standard JSON values such as `NaN`, positive or negative infinity, and + Python objects MUST be rejected. +- All timestamps use RFC 3339 UTC and end in `Z`, for example + `2026-08-23T08:00:01.420Z`. +- `duration_ms` is a non-negative number in milliseconds. It is computed from + a monotonic clock interval and is not used for event ordering. + +## Journal Entry envelope + +Each JSONL line has this top-level shape: + +```json +{ + "schema_version": 1, + "run_id": "run-7d9e81d0-2dbe-4d4c-a473-62582e5dc842", + "seq": 4, + "recorded_at": "2026-08-23T08:00:01.420Z", + "type": "tool.completed", + "payload": { + "model_call_id": "model-bb7241c2-348d-4bb6-975a-b33f05ce76b2", + "tool_call_id": "toolu_01Abc", + "tool_name": "read", + "result": "file contents", + "is_error": false, + "duration_ms": 3.72, + "source_timestamp": "2026-08-23T08:00:01.419Z" + } +} +``` + +| Field | Type | Meaning | +|---|---|---| +| `schema_version` | integer | The Journal Entry wire-schema version. Fixed at `1` in v1. | +| `run_id` | non-empty string | The Agent Run identity. All entries in one file MUST match. | +| `seq` | positive integer | The authoritative order within a run. The writer begins at `1` and increments once per entry. | +| `recorded_at` | RFC 3339 UTC string | Wall-clock time when the Journal writer accepted and recorded the fact. It is not the ordering key. | +| `type` | string | Native Event type. v1 accepts only the types in this document's event catalog. | +| `payload` | object | Facts belonging to the event. | +| `truncation` | object, optional | Present only when persistence truncated a string. See “Truncation protocol.” | + +`seq` is the only authoritative ordering key within a run. `recorded_at` values +may be equal or affected by wall-clock adjustments, so consumers MUST NOT +reorder entries solely by timestamp. + +## Source-time field shared by all events + +Every v1 event payload contains: + +| Field | Required | Type | Meaning | +|---|---:|---|---| +| `source_timestamp` | yes | RFC 3339 UTC string or `null` | The fact occurrence time known by the Native Event producer. nano core timestamps the event boundary; an adapter copies only trustworthy time from an upstream Source Record; when no such time is known it writes `null`. | + +`source_timestamp` belongs to the Native Event, while `recorded_at` belongs to +the Journal Entry. They describe different stages and cannot substitute for one +another. + +Here producer means the component that creates the Native Event, not the model +provider: + +| Producer case | `source_timestamp` | +|---|---| +| Current nano core produces the event directly | Core reads UTC at the corresponding user, model, tool, or run event boundary. | +| A future adapter normalizes a Source Record with trustworthy time | The adapter copies the upstream time; it does not replace it with adapter receive time. | +| A future adapter receives a Source Record without trustworthy time | It writes `null`; the Journal writer still preserves receive time in `recorded_at`. | + +Native Event v1 explicitly rejects `timestamp_source`. A future ATIF projector +may record whether it ultimately selected `source_timestamp` or `recorded_at` +in ATIF `extra`, but that is a projection decision rather than a Native Event +runtime fact. + +## Event catalog + +v1 supports nine event types: + +| Event | Semantics | +|---|---| +| `run.started` | The Agent Run exists and its execution parameters are fixed. | +| `user.message` | The entry user message for this Agent Run. | +| `model.started` | One model call has begun. | +| `model.output_delta` | The model streamed one text fragment. | +| `model.completed` | One model call completed successfully and its final message and usage are available. | +| `tool.started` | One tool call began. | +| `tool.completed` | One tool call ended with a normal result, a tool-level error, or an exception. | +| `run.completed` | The Agent Run ended normally, including turn-budget exhaustion. | +| `run.failed` | The Agent Run failed because of an unhandled exception. | + +### `run.started` + +| Field | Type | Meaning | +|---|---|---| +| `mode` | `"interactive"` or `"headless"` | The run mode. | +| `model` | non-empty string | Requested model identifier. | +| `max_turns` | positive integer or `null` | Maximum number of model-call turns; currently `null` in interactive mode. | +| `producer` | object | Identity of the program producing Native Events. It MUST contain non-empty string fields `name` and `version`. nano core writes `{ "name": "nanoPyCodeAgent", "version": }`. | + +This MUST be the first event produced by nano core. It says that the Agent Run +has begun, not that a model request has already been sent. + +`producer.version` comes from installed package metadata. Development versions +built by hatch-vcs normally include a Git revision. When running directly from +an uninstalled source tree and package metadata is unavailable, core writes +`"unknown"`. `producer` is run-level provenance and is not subject to the +string-size truncation limit. + +`producer.version` is orthogonal to the Journal Entry `schema_version`: the +former answers which nanoPyCodeAgent build produced the run; the latter tells a +reader which wire schema parses each entry. Consumers MUST NOT infer either one +from the other. + +### `user.message` + +| Field | Type | Meaning | +|---|---|---| +| `message_id` | non-empty string | Local identity generated by nano for the entry user message. | +| `content` | any JSON value | User content for this run. The current CLI produces a string; the protocol permits structured content. | + +In interactive mode, each user input creates a new Agent Run and Journal. Prior +conversation remains in process as model context, but the new Journal does not +copy it as a complete request snapshot and v1 has no session link. + +### `model.started` + +| Field | Type | Meaning | +|---|---|---| +| `model_call_id` | non-empty string | Local correlation ID generated by nano for one model call. | +| `model` | non-empty string | Model identifier requested for this call. | + +An Agent Run may contain multiple model calls. Each one receives a new +`model_call_id`. + +### `model.output_delta` + +| Field | Type | Meaning | +|---|---|---| +| `model_call_id` | non-empty string | The associated model call. | +| `delta` | string | Text added by this streaming callback; it may be empty. | + +This event represents text deltas only. A tool-only response may have no delta. +The complete text also appears in a text block in the following +`model.completed.content`. This intentional duplication preserves both the +real-time process and the final completed state. + +### `model.completed` + +| Field | Type | Meaning | +|---|---|---| +| `model_call_id` | non-empty string | Associated local model-call ID. | +| `message_id` | non-empty string | Stable identity of the completed message; the provider response ID is preferred, falling back to `model_call_id`. | +| `content` | array | Complete provider-neutral message blocks; see the schema below. | +| `tool_calls` | array | Ordered copies of all `tool_call` blocks in `content`; they MUST match item by item. | +| `model` | non-empty string | Actual model returned by the provider, falling back to the requested model. | +| `stop_reason` | string or `null` | Provider stop reason, such as `end_turn` or `tool_use`. | +| `usage` | object or `null` | Token usage for this model call; see the schema below. | +| `provider_response_id` | non-empty string or `null` | Original provider response/message ID. | +| `generation_id` | non-empty string or `null` | Provider generation ID, currently read from the `x-generation-id` response header. | +| `duration_ms` | non-negative number | Time from starting the request until the complete message and response headers are available. | + +`content` supports these blocks: + +| `type` | Other fields | Meaning | +|---|---|---| +| `text` | `text: string` | A complete text fragment. | +| `tool_call` | `tool_call_id: non-empty string`, `tool_name: non-empty string`, `input: object` | A provider-neutral tool call. | +| `extension` | `namespace: non-empty string`, `source_type: string \| null`, `value: JSON value` | A provider block that has not been normalized. nano's Anthropic adapter uses `namespace: "anthropic"`. | + +An unknown provider block MUST be wrapped in `extension`; it must not be +silently dropped or represented by inventing another `content.type`. + +When `usage` is not `null`: + +| Field | Required | Type | Meaning | +|---|---:|---|---| +| `input_tokens` | yes | non-negative integer | Input tokens reported by the provider. | +| `output_tokens` | yes | non-negative integer | Output tokens reported by the provider. | +| `cache_read_input_tokens` | no | non-negative integer | Input tokens read from the prompt cache. | +| `cache_creation_input_tokens` | no | non-negative integer | Input tokens written to the prompt cache. | + +Other JSON usage fields returned by the provider MAY be preserved. v1 does not +derive cost from usage or a price catalog. + +### `tool.started` + +| Field | Required | Type | Meaning | +|---|---:|---|---| +| `tool_call_id` | yes | non-empty string | Provider tool-call ID, matching `model.completed.tool_calls[].tool_call_id`. | +| `tool_name` | yes | non-empty string | Tool name. | +| `input` | yes | object | Complete tool input. | +| `model_call_id` | core profile | non-empty string | Model call that produced this tool call. nano core always writes it; the base v1 payload validator permits omission for normalized sources. | + +The same tool call appears in both `model.completed.tool_calls` and the tool +lifecycle events. The former preserves the model action; the latter preserves +the actual execution boundary. + +### `tool.completed` + +| Field | Required | Type | Meaning | +|---|---:|---|---| +| `tool_call_id` | yes | non-empty string | Same value as the corresponding `tool.started`. | +| `tool_name` | yes | non-empty string | Tool name. | +| `result` | yes | string or `null` | Text returned to the model; `null` when execution raises an exception. | +| `is_error` | yes | boolean | Whether the result represents an error. An expected tool failure may still have a string result and set this to `true`. | +| `duration_ms` | yes | non-negative number | Tool execution duration. | +| `error` | conditionally | object | Required when `result` is `null`. nano core writes `{ "type": ..., "message": ... }`. | +| `model_call_id` | core profile | non-empty string | Model call that produced this tool call; nano core always writes it. | + +An expected tool error ends with `tool.completed`, and the agent may continue by +sending the result back to the model. An unhandled tool exception first produces +`tool.completed` with `result: null` and `is_error: true`, then causes the run to +produce `run.failed`. + +### `run.completed` + +| Field | Type | Meaning | +|---|---|---| +| `outcome` | `"completed"` or `"max_turns_exhausted"` | Normal completion reason. Turn-budget exhaustion is an explainable terminal state, not an exception. | +| `duration_ms` | non-negative number | Total Agent Run duration. | + +When the final model reply still requests tools but `max_turns` is exhausted, +core does not execute those tools and directly records `max_turns_exhausted`. + +### `run.failed` + +| Field | Type | Meaning | +|---|---|---| +| `error_type` | non-empty string | Python type name of the unhandled exception. | +| `message` | string | Exception message; it may be empty. | +| `duration_ms` | non-negative number | Time from run start until failure. | + +After `run.failed` is recorded, the original exception continues to propagate to +the caller. CLI argument errors, settings-loading failures, and missing API +credentials happen before an Agent Run is established, so they have no Journal +and do not produce `run.failed`. + +## nano core event ordering + +Core currently guarantees these typical sequences: + +```text +# Successful run without tools +run.started +user.message +model.started +model.output_delta * +model.completed +run.completed(outcome = completed) + +# Successful run with tools +run.started +user.message +model.started +model.output_delta * +model.completed(stop_reason = tool_use) +(tool.started → tool.completed) * +model.started +... +run.completed(outcome = completed) + +# Model or runtime exception +run.started +user.message +... +run.failed +``` + +Here `*` means zero or more occurrences. A run MUST end with exactly one of +`run.completed` or `run.failed`; a failed model call does not produce +`model.completed`. + +These are state-machine guarantees of the nano core producer. v1 `replay()` +currently validates only each entry's schema, a single `run_id`, and strictly +increasing `seq`; it does not perform cross-event state-machine validation. +Consumers must not equate “the file can be replayed” with “the lifecycle is +complete.” A forcibly terminated process may have no terminal event. + +## Identity and correlation rules + +- `run_id` currently has the form `run-` and determines the filename. +- `message_id` identifies a complete user or model message. +- `model_call_id` correlates `model.started`, all its deltas, + `model.completed`, and tool events triggered by that reply. +- `tool_call_id` correlates the model action, `tool.started`, and + `tool.completed`. +- Local IDs need only be stable and non-empty in their applicable scope. + Consumers SHOULD NOT parse UUID formatting for semantics. + +## Truncation protocol + +A Journal may contain very large model text, tool input, and tool output. The +writer independently limits every string in the payload, by default to `100000` +Unicode code points: + +- Only the persisted copy keeps the string prefix. +- The original Native Event and live stdout projector remain untruncated. +- A truncated Journal Entry gains a top-level `truncation` object: + +```json +{ + "fields": [ + { + "path": "/result", + "original_chars": 150000, + "retained_chars": 100000 + } + ] +} +``` + +`path` is a JSON Pointer rooted at `payload`. Array indexes use decimal notation; +`~` and `/` in object keys are escaped as `~0` and `~1`. `original_chars` and +`retained_chars` count Python Unicode characters, not UTF-8 bytes. + +The following identity, classification, and time metadata fields are not +truncated, preserving correlation and schema validity: + +```text +error_type, generation_id, message_id, mode, model, model_call_id, +outcome, producer, provider_response_id, source_timestamp, stop_reason, +tool_call_id, tool_name +``` + +Truncation means that the Journal has lost the tail of that field. Consumers +MUST NOT treat the retained prefix as a complete value. + +## Storage and append semantics + +The default location is: + +```text +~/.nanoPyCodeAgent/journals/.jsonl +``` + +The protocol and implementation impose these constraints: + +- One Agent Run corresponds to one file. +- `run_id` MUST match `[A-Za-z0-9][A-Za-z0-9._-]{0,127}`, preventing directory + traversal. +- The writer opens files in create-exclusive, append-only mode. It fails rather + than overwriting an existing file. +- On supported platforms, close-on-exec and no-follow flags are also used. +- The configuration root and Journal directory are set to mode `0700`; files + are set to mode `0600`. +- A lock serializes append and close operations within one `EventJournal` + instance. +- A write loop completes each record, and the next `seq` advances only after a + complete write. +- A normal close calls `fsync` before closing the descriptor. + +v1 provides no automatic rotation, retention policy, encryption, compression, +cross-process writer coordination, or Journal management CLI. + +## Replay behavior + +`EventJournal.replay(path)` returns Journal Entries in file order and enforces: + +- Every newline-terminated record MUST be a valid UTF-8 JSON object. +- `schema_version` MUST be the reader-supported value `1`. +- The entry envelope and Native Event payload MUST pass v1 validation. +- All complete entries MUST have the same `run_id`. +- `seq` MUST be a positive integer and strictly increasing. The reader permits + gaps, while the writer normally produces `1, 2, 3, ...` continuously. +- A final fragment without a newline is treated as a partial tail left by an + interrupted final write and is ignored. +- A corrupt line in the middle of the file, or any invalid newline-terminated + final line, MUST raise an error and cannot be skipped. + +Replay does not repair the file or validate a state machine, digest, signature, +or tamper-evident chain. “Replayable” means that completely written, +schema-valid facts can be recovered; it does not make the Event Journal a +transactional database or trusted audit log. + +## Sensitive information and data scope + +The Journal explicitly records: + +- the current user input; +- complete model output, streaming text, and tool calls; +- complete tool input and the tool result returned to the model; +- provider message/generation IDs, stop reason, and usage; +- the program name and package version that produced the run; and +- error type, error message, and stage durations. + +It may therefore contain source code, paths, credentials, or other secrets +indirectly through prompts, model output, shell commands, file content, or tool +results. Modes `0700` and `0600` provide only minimum local access control; they +do not provide redaction, encryption, or secret scanning. Journals MUST NOT be +uploaded, published, or shared as ordinary diagnostic attachments by default. + +v1 does not explicitly record: + +- API keys or authentication headers; +- complete provider requests, HTTP headers, or raw SDK responses; +- the system prompt or a complete history snapshot sent to the model; +- stdout presentation details such as spinners, ANSI color, prompts, or banners; +- token cost or price-catalog resolution; +- session identity or parent-child relationships across runs; or +- an ATIF trajectory or public `stream-json` record. + +“No explicit field” does not mean that equivalent data cannot appear inside +content. For example, a secret placed in a shell command is still recorded in +`tool.started.input`. + +## stdout and public-interface boundary + +When v1 introduced the Event Journal, the existing stdout text had to remain +byte-for-byte behaviorally unchanged. The current `_TextOutputProjector` +consumes only: + +- `model.output_delta` to print the reply prefix and streaming text; +- `model.completed` to add a newline after a response that emitted text; +- `tool.started` to print the tool-call preview; and +- `tool.completed` to print a string result. + +The Journal path, run ID, recording timestamp, and other envelope metadata are +not written to stdout. The Event Journal is internal reconstruction data, not a +stable user-output contract. External programs SHOULD NOT treat +`~/.nanoPyCodeAgent/journals/*.jsonl` as a public CLI API. + +## Versioning and compatibility + +A v1 reader fails closed on an unknown `schema_version` or event type. The +compatibility rules are: + +`schema_version` governs protocol compatibility, while +`run.started.producer.version` provides producer provenance only. A producer +implementation fix that leaves the wire contract unchanged changes only the +package version, not the schema. A required field, type, or semantic change +increments the schema according to the rules below. + +The required `producer` field was finalized before the first merge and release +of v1, so it belongs to the initial v1 contract and is not a compatibility +change to a released protocol. The following rules govern evolution after v1 +is released. + +- An optional payload field that does not change existing meaning MAY be added + while retaining `schema_version = 1`; an old reader ignores semantics it does + not understand. +- Provider-specific model content SHOULD use an `extension` block instead of a + new block type. +- Adding an event type or required field, changing a field type or meaning, or + changing the envelope or ordering rules MUST increment `schema_version`. +- A version increment MUST update both language versions of this protocol, the + producer, reader/replay implementation, and contract tests together. +- Unknown top-level fields are currently ignored by the reader, but are not + guaranteed to survive deserialize/serialize. Extensions SHOULD live in a + payload or `extension` block with explicit ownership. + +Event Journal v1 is an internal protocol and does not promise permanent +compatibility across nanoPyCodeAgent major versions. `schema_version` exists so +that incompatible changes are rejected explicitly rather than silently +misinterpreted. + +## Implementation map + +| Behavior | Location | +|---|---| +| Event types, payload validation, and content/usage schemas | [`event_journal.py`](../../../src/nanopycodeagent/event_journal.py) | +| Journal Entry encoding, truncation, permissions, append, fsync, and replay | [`event_journal.py`](../../../src/nanopycodeagent/event_journal.py) | +| Run, user, model, and tool event emission | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| Anthropic block normalization into provider-neutral content | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| Native Event projection into existing stdout text | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| Envelope, ordering, permissions, truncation, partial-tail, and schema tests | [`test_event_journal.py`](../../../tests/test_event_journal.py) | +| Event lifecycle and exact stdout-projection regression tests | [`test_agent_events.py`](../../../tests/test_agent_events.py) | +| Existing agent-loop and stdout behavior tests | [`test_agent.py`](../../../tests/test_agent.py) | diff --git a/docs/dev_docs/zh-CN/event-journal-protocol-v1.md b/docs/dev_docs/zh-CN/event-journal-protocol-v1.md new file mode 100644 index 0000000..8c963fd --- /dev/null +++ b/docs/dev_docs/zh-CN/event-journal-protocol-v1.md @@ -0,0 +1,431 @@ +# Event Journal 实现协议 v1 + +> 本文件为**中文源文件**（source of truth）；英文版 +> [`../en/event-journal-protocol-v1.md`](../en/event-journal-protocol-v1.md) +> 由其生成。 + +| 项目 | 值 | +|---|---| +| 状态 | 已实现（Implemented） | +| 协议版本 | v1 | +| `schema_version` | `1` | +| 可见性 | 内部协议；不是公开 Run Output 或 ATIF 接口 | +| 领域术语 | [`CONTEXT.md`](../../../CONTEXT.md) | +| 核心实现 | [`event_journal.py`](../../../src/nanopycodeagent/event_journal.py) | +| 事件生产与文本投影 | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| 行为测试 | [`test_event_journal.py`](../../../tests/test_event_journal.py)、[`test_agent_events.py`](../../../tests/test_agent_events.py)、[`test_agent.py`](../../../tests/test_agent.py) | + +## 文档定位 + +这是一份**已实现协议规范**，不是 RFC。 + +- RFC 用于实现前或行为变更前的提案、讨论和评审。 +- ADR 用于记录难以逆转、反直觉或存在重要取舍的架构决策及其理由。 +- 本文固定当前已经落地的 wire contract、事件语义、持久化行为和兼容性边界。 + +如果后续要改变这些行为，可以先写 RFC 或 ADR；变更被接受并实现后，再更新本文和协议版本。 + +本文中的“必须”“应当”“可以”分别表示协议要求、推荐行为和允许行为。除非特别说明，字段和校验规则描述的是 `schema_version = 1`。 + +## 目标与边界 + +Event Journal 为一次 **Agent Run** 保存可重放的内部运行事实。完整链路是： + +```text +Agent core + ↓ 产生 +Native Event {type, payload} + ├─→ live projector → 现有 stdout 文本 Run Output + └─→ journal writer → Journal Entry + ↓ UTF-8 JSONL，追加写 + Event Journal + ↓ 后续 projector（不属于 v1） + ATIF Trajectory / 其他公开表示 +``` + +这里的边界是： + +- **Native Event** 是 agent-independent 的运行事实。 +- **Journal Entry** 为 Native Event 增加持久化所需的身份、顺序和记录时间。 +- **Event Journal** 是单个 Agent Run 的 Journal Entry 追加序列。 +- stdout 是同一组 Native Event 的实时文本投影，但不等于 Event Journal。 +- ATIF、`stream-json` 和其他公开输出是后续投影，不属于 v1。 + +Event emitter 必须先把事件追加到 Journal，再交给 live projector。持久化截断只作用于 Journal Entry；projector 收到的是未截断的 Native Event，因此现有 stdout 行为不受 Journal 大小限制影响。 + +## 编码与基本类型 + +- 文件使用 UTF-8 JSONL；每个完整行只包含一个 Journal Entry，并以 `\n` 结束。 +- writer 使用紧凑 JSON 编码，不要求字段顺序具有语义。 +- payload 必须是 JSON object；其递归值只允许 `null`、boolean、有限 number、string、array 和 object。 +- `NaN`、正负无穷、Python 对象等非标准 JSON 值必须被拒绝。 +- 所有时间戳使用 RFC 3339 UTC 格式，并以 `Z` 结尾，例如 `2026-08-23T08:00:01.420Z`。 +- `duration_ms` 是非负 number，单位为毫秒；它来自 monotonic clock 的耗时差，不用于事件排序。 + +## Journal Entry envelope + +每个 JSONL 行的顶层结构如下： + +```json +{ + "schema_version": 1, + "run_id": "run-7d9e81d0-2dbe-4d4c-a473-62582e5dc842", + "seq": 4, + "recorded_at": "2026-08-23T08:00:01.420Z", + "type": "tool.completed", + "payload": { + "model_call_id": "model-bb7241c2-348d-4bb6-975a-b33f05ce76b2", + "tool_call_id": "toolu_01Abc", + "tool_name": "read", + "result": "file contents", + "is_error": false, + "duration_ms": 3.72, + "source_timestamp": "2026-08-23T08:00:01.419Z" + } +} +``` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `schema_version` | integer | Journal Entry wire schema 版本。v1 固定为 `1`。 | +| `run_id` | non-empty string | Agent Run 的身份；同一文件中的所有 entry 必须相同。 | +| `seq` | positive integer | run 内权威顺序号。writer 从 `1` 开始逐条递增。 | +| `recorded_at` | RFC 3339 UTC string | Journal writer 接受并记录该事实的墙上时间。它不是排序依据。 | +| `type` | string | Native Event 类型。v1 只允许本文事件目录中的类型。 | +| `payload` | object | 对应事件的事实数据。 | +| `truncation` | object，可选 | 仅当持久化时有字符串被截断才出现；详见“截断协议”。 | + +`seq` 是 run 内排序的唯一权威。`recorded_at` 可能相同，也可能受墙上时钟调整影响，消费者不得仅按时间戳重排。 + +## 所有事件共有的来源时间字段 + +每种 v1 事件的 payload 都包含： + +| 字段 | 必需 | 类型 | 含义 | +|---|---:|---|---| +| `source_timestamp` | 是 | RFC 3339 UTC string 或 `null` | Native Event producer 所知道的事实发生时间。nano core 在事件边界采时；adapter 只复制上游 Source Record 的可信时间；无法确定时写 `null`。 | + +`source_timestamp` 属于 Native Event，`recorded_at` 属于 Journal Entry。两者表达不同阶段，不得互相替代。 + +这里的 producer 是产生 Native Event 的组件，不是模型供应商： + +| producer 情况 | `source_timestamp` | +|---|---| +| 当前 nano core 直接产生事件 | core 在 user、model、tool 或 run 的对应事件边界读取 UTC。 | +| 未来 adapter 从带可信时间的 Source Record 归一化事件 | 复制上游时间，不改写成 adapter 的接收时间。 | +| 未来 adapter 的 Source Record 没有可信时间 | 写 `null`；Journal writer 的接收时间仍由 `recorded_at` 保存。 | + +Native Event v1 明确拒绝 `timestamp_source`。未来 ATIF projector 可以在 ATIF `extra` 中记录它最终选择了 `source_timestamp` 还是 `recorded_at`，但这属于投影决策，不是 Native Event 运行事实。 + +## 事件目录 + +v1 支持九种事件： + +| 事件 | 语义 | +|---|---| +| `run.started` | Agent Run 已建立，运行参数已经确定。 | +| `user.message` | 本次 Agent Run 的入口用户消息。 | +| `model.started` | 一次模型调用开始。 | +| `model.output_delta` | 模型流式产生一段文本。 | +| `model.completed` | 一次模型调用成功完成，最终消息和 usage 已可用。 | +| `tool.started` | 一个工具调用开始。 | +| `tool.completed` | 一个工具调用以正常结果、工具级错误或异常结束。 | +| `run.completed` | Agent Run 正常结束，包括达到轮次上限。 | +| `run.failed` | Agent Run 因未处理异常失败。 | + +### `run.started` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `mode` | `"interactive"` 或 `"headless"` | 本次运行模式。 | +| `model` | non-empty string | 请求使用的模型标识。 | +| `max_turns` | positive integer 或 `null` | 最大模型调用轮数；interactive 当前为 `null`。 | +| `producer` | object | 产生 Native Event 的程序身份；必须包含 non-empty string `name` 和 `version`。nano core 写入 `{ "name": "nanoPyCodeAgent", "version": }`。 | + +该事件必须是 nano core 产生的第一个事件。它只说明 Agent Run 已开始，不表示模型请求已经发出。 + +`producer.version` 来自安装包元数据。hatch-vcs 构建的开发版本通常包含 Git revision；直接从未安装的源码运行、无法读取包元数据时写 `"unknown"`。`producer` 是 run 级溯源信息，不随字符串大小上限截断。 + +`producer.version` 与 Journal Entry 的 `schema_version` 正交：前者回答“哪个 nanoPyCodeAgent 构建产生了这次运行”，后者回答“reader 应按哪个 wire schema 解析每条记录”。消费者不得用其中一个推断另一个。 + +### `user.message` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `message_id` | non-empty string | nano 为入口用户消息生成的本地身份。 | +| `content` | 任意 JSON value | 本次输入的用户内容。当前 CLI 产生 string；协议允许结构化内容。 | + +interactive 模式下，每次用户输入建立一个新的 Agent Run 和 Journal。此前对话仍会作为模型上下文保存在进程内，但不会在新 Journal 中复制成完整请求快照，也没有 v1 session link。 + +### `model.started` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `model_call_id` | non-empty string | nano 为一次模型调用生成的本地关联 ID。 | +| `model` | non-empty string | 本次请求使用的模型标识。 | + +一个 Agent Run 可以有多次模型调用；每次都使用新的 `model_call_id`。 + +### `model.output_delta` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `model_call_id` | non-empty string | 关联的模型调用。 | +| `delta` | string | 本次流式回调新增的文本，可为空字符串。 | + +该事件只表示文本增量。纯工具调用回复可以没有任何 delta。完整文本仍会出现在随后 `model.completed.content` 的 text block 中；这项有意的重复同时保留实时过程和最终完成态。 + +### `model.completed` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `model_call_id` | non-empty string | 关联的本地模型调用 ID。 | +| `message_id` | non-empty string | 完成消息的稳定身份；优先使用 provider response ID，缺失时回退到 `model_call_id`。 | +| `content` | array | provider-neutral 的完整消息 block；schema 见下文。 | +| `tool_calls` | array | `content` 中全部 `tool_call` block 的有序副本，必须逐项完全相等。 | +| `model` | non-empty string | provider 返回的实际模型；缺失时回退到请求模型。 | +| `stop_reason` | string 或 `null` | provider 的停止原因，例如 `end_turn` 或 `tool_use`。 | +| `usage` | object 或 `null` | 本次模型调用的 token usage；schema 见下文。 | +| `provider_response_id` | non-empty string 或 `null` | provider 原始 response/message ID。 | +| `generation_id` | non-empty string 或 `null` | provider generation ID；当前从 `x-generation-id` response header 读取。 | +| `duration_ms` | non-negative number | 从开始请求到完整消息和响应头可用的耗时。 | + +`content` 支持以下 block： + +| `type` | 其他字段 | 含义 | +|---|---|---| +| `text` | `text: string` | 完整文本片段。 | +| `tool_call` | `tool_call_id: non-empty string`、`tool_name: non-empty string`、`input: object` | provider-neutral 工具调用。 | +| `extension` | `namespace: non-empty string`、`source_type: string \| null`、`value: JSON value` | 尚未标准化的 provider block。nano 的 Anthropic adapter 使用 `namespace: "anthropic"`。 | + +未知 provider block 必须包装成 `extension`，不能静默丢弃，也不能直接发明新的 `content.type`。 + +当 `usage` 非 `null` 时： + +| 字段 | 必需 | 类型 | 含义 | +|---|---:|---|---| +| `input_tokens` | 是 | non-negative integer | provider 报告的输入 token 数。 | +| `output_tokens` | 是 | non-negative integer | provider 报告的输出 token 数。 | +| `cache_read_input_tokens` | 否 | non-negative integer | 从 prompt cache 读取的输入 token 数。 | +| `cache_creation_input_tokens` | 否 | non-negative integer | 写入 prompt cache 的输入 token 数。 | + +provider 返回的其他 JSON usage 字段可以原样保留。v1 不根据 usage 或价格目录计算 cost。 + +### `tool.started` + +| 字段 | 必需 | 类型 | 含义 | +|---|---:|---|---| +| `tool_call_id` | 是 | non-empty string | provider 工具调用 ID；与 `model.completed.tool_calls[].tool_call_id` 对应。 | +| `tool_name` | 是 | non-empty string | 工具名称。 | +| `input` | 是 | object | 完整的工具输入。 | +| `model_call_id` | core profile | non-empty string | 产生该工具调用的模型调用。nano core 总是写入；基础 v1 payload validator 为兼容归一化来源允许省略。 | + +同一个工具调用会同时出现在 `model.completed.tool_calls` 和工具生命周期事件中：前者保存模型的 action，后者保存实际执行边界。 + +### `tool.completed` + +| 字段 | 必需 | 类型 | 含义 | +|---|---:|---|---| +| `tool_call_id` | 是 | non-empty string | 与对应 `tool.started` 相同。 | +| `tool_name` | 是 | non-empty string | 工具名称。 | +| `result` | 是 | string 或 `null` | 工具返回给模型的文本；执行抛出异常时为 `null`。 | +| `is_error` | 是 | boolean | 结果是否表示错误。工具的预期失败也可产生 string result 并设为 `true`。 | +| `duration_ms` | 是 | non-negative number | 工具执行耗时。 | +| `error` | 条件必需 | object | `result` 为 `null` 时必须存在。nano core 写 `{ "type": ..., "message": ... }`。 | +| `model_call_id` | core profile | non-empty string | 产生该工具调用的模型调用；nano core 总是写入。 | + +预期内的工具错误会结束于 `tool.completed`，agent 可以继续把结果交给模型。工具抛出的未处理异常会先产生 `tool.completed`（`result: null`、`is_error: true`），再使 run 产生 `run.failed`。 + +### `run.completed` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `outcome` | `"completed"` 或 `"max_turns_exhausted"` | 正常结束原因。达到轮次上限属于可解释的正常终态，不是异常。 | +| `duration_ms` | non-negative number | 整个 Agent Run 的耗时。 | + +当最后一轮模型回复仍请求工具、但 `max_turns` 已耗尽时，core 不执行这些工具，直接记录 `max_turns_exhausted`。 + +### `run.failed` + +| 字段 | 类型 | 含义 | +|---|---|---| +| `error_type` | non-empty string | 未处理异常的 Python 类型名。 | +| `message` | string | 异常消息，可以为空。 | +| `duration_ms` | non-negative number | 从 run 开始到失败的耗时。 | + +`run.failed` 记录后，原异常继续向调用方传播。CLI 参数错误、设置加载失败和缺少 API credential 发生在 Agent Run 建立之前，因此没有 Journal，也不会产生 `run.failed`。 + +## nano core 事件顺序 + +core 当前保证以下典型序列： + +```text +# 无工具的成功运行 +run.started +user.message +model.started +model.output_delta * +model.completed +run.completed(outcome = completed) + +# 含工具的成功运行 +run.started +user.message +model.started +model.output_delta * +model.completed(stop_reason = tool_use) +(tool.started → tool.completed) * +model.started +... +run.completed(outcome = completed) + +# 模型或运行时异常 +run.started +user.message +... +run.failed +``` + +其中 `*` 表示零次或多次。一个 run 必须以且只以 `run.completed` 或 `run.failed` 结束；失败的模型调用不会产生 `model.completed`。 + +这些是 nano core producer 的状态机保证。v1 `replay()` 当前只校验每条 entry 的 schema、单一 `run_id` 和严格递增的 `seq`，不执行跨事件状态机校验。消费者不能把“文件可 replay”误解为“生命周期一定完整”；进程被强制终止时可能没有终态事件。 + +## 身份与关联规则 + +- `run_id` 当前格式为 `run-`，并决定文件名。 +- `message_id` 标识完整用户或模型消息。 +- `model_call_id` 关联 `model.started`、其所有 delta、`model.completed` 和由该回复触发的工具事件。 +- `tool_call_id` 关联模型 action、`tool.started` 和 `tool.completed`。 +- 所有本地 ID 只要求在相应作用域内稳定且非空；消费者不应解析 UUID 格式获取语义。 + +## 截断协议 + +Journal 可能包含超大模型文本、工具输入和工具结果。writer 对 payload 中的每个 string 独立执行字符数上限，默认 `100000` 个 Unicode code point： + +- 只在持久化副本中保留字符串前缀。 +- 原始 Native Event 及 live stdout projector 不截断。 +- 被截断的 Journal Entry 增加顶层 `truncation`： + +```json +{ + "fields": [ + { + "path": "/result", + "original_chars": 150000, + "retained_chars": 100000 + } + ] +} +``` + +`path` 是以 `payload` 为根的 JSON Pointer。数组索引按十进制表示，object key 中的 `~` 和 `/` 分别转义为 `~0` 和 `~1`。`original_chars` 和 `retained_chars` 按 Python Unicode 字符数计，不是 UTF-8 byte 数。 + +以下身份、分类和时间元数据字段不截断，以保持关联和 schema 有效： + +```text +error_type, generation_id, message_id, mode, model, model_call_id, +outcome, producer, provider_response_id, source_timestamp, stop_reason, +tool_call_id, tool_name +``` + +截断表示 Journal 已丢失该字段的尾部，消费者不得把保留前缀当作完整值。 + +## 存储与追加语义 + +默认位置： + +```text +~/.nanoPyCodeAgent/journals/.jsonl +``` + +协议和实现约束如下： + +- 一个 Agent Run 对应一个文件。 +- `run_id` 必须匹配 `[A-Za-z0-9][A-Za-z0-9._-]{0,127}`，防止目录逃逸。 +- writer 以 create-exclusive、append-only 方式打开文件；已有同名文件时失败，不覆盖。 +- 支持的平台上同时使用 close-on-exec 和 no-follow 标志。 +- 配置根目录和 Journal 目录权限设为 `0700`，文件权限设为 `0600`。 +- 同一个 `EventJournal` 实例内的 append 和 close 由 lock 串行化。 +- 每条记录通过 write loop 写完；只有完整写入后才递增下一个 `seq`。 +- 正常 close 会先 `fsync` 再关闭 descriptor。 + +v1 不提供自动轮转、保留期限、加密、压缩、跨进程 writer 协调或 Journal 管理 CLI。 + +## Replay 行为 + +`EventJournal.replay(path)` 按文件顺序返回 Journal Entry，并执行： + +- 每个换行结束的记录必须是合法 UTF-8 JSON object。 +- `schema_version` 必须为 reader 支持的 `1`。 +- entry envelope 和 Native Event payload 必须通过 v1 校验。 +- 所有完整 entry 的 `run_id` 必须相同。 +- `seq` 必须为正整数且严格递增；reader 容许 gap，但 writer 正常情况下连续产生 `1, 2, 3, ...`。 +- 最后一个没有换行的片段被视为进程在最终 write 中断留下的 partial tail，并被忽略。 +- 位于文件中间的损坏行或任何已换行的无效尾行必须报错，不能跳过。 + +Replay 不修复文件，也不验证事件状态机、摘要、签名或防篡改链。Event Journal 的“可重放”表示可以恢复已完整写入并通过 schema 校验的事实，不表示它是事务数据库或可信审计日志。 + +## 敏感信息与数据范围 + +Journal 明确记录： + +- 本次用户输入； +- 完整模型输出、流式文本和工具调用； +- 完整工具输入和返回给模型的工具结果； +- provider message/generation ID、stop reason 和 usage； +- 产生本次运行的程序名称和包版本； +- 错误类型、错误消息和各阶段耗时。 + +因此它可能通过 prompt、模型输出、shell command、文件内容或工具结果间接包含源码、路径、credential 或其他秘密。`0700`/`0600` 只是本机最小访问控制，不等于脱敏、加密或秘密扫描。不得默认上传、公开或作为普通诊断附件分享 Journal。 + +v1 没有专门记录： + +- API key 或 auth header； +- 完整 provider request、HTTP header 或 SDK 原始 response； +- system prompt 和发给模型的完整历史快照； +- spinner、ANSI 颜色、提示符、banner 等 stdout 表现细节； +- token cost 或价格目录解析结果； +- session 身份、跨 run 父子关系； +- ATIF trajectory 或 public `stream-json` 记录。 + +“没有专门字段”不代表内容中不可能出现同类数据；例如用户把 secret 写进 shell command 时，它仍会随 `tool.started.input` 被记录。 + +## stdout 与公开接口边界 + +v1 引入 Event Journal 时，现有 stdout 文本必须保持完全不变。当前 `_TextOutputProjector` 只消费： + +- `model.output_delta`：输出 reply prefix 和流式文本； +- `model.completed`：在已经输出文本时补换行； +- `tool.started`：输出工具调用预览； +- `tool.completed`：输出 string result。 + +Journal file path、run ID、recorded timestamp 和其他 envelope 元数据不写到 stdout。Event Journal 是内部 reconstruction data，不是稳定的用户输出 contract；外部程序不应把 `~/.nanoPyCodeAgent/journals/*.jsonl` 当作公共 CLI API。 + +## 版本与兼容性 + +v1 reader 对未知 `schema_version` 和未知事件类型 fail closed。兼容性规则是： + +`schema_version` 管理协议兼容性，`run.started.producer.version` 只提供生产者溯源。修复 producer 实现但不改变 wire contract 时，只改变包版本，不提升 schema；改变必需字段、类型或语义时才按以下规则提升 schema。 + +`producer` 必需字段在 v1 首次合并和发布前完成，因此属于初始 v1 contract，不构成已发布协议的兼容性变更。以下规则适用于 v1 发布后的演进。 + +- 增加不改变既有含义的可选 payload 字段，可以保持 `schema_version = 1`；旧 reader 会忽略自己不理解的附加语义。 +- provider-specific model content 应优先放进 `extension` block，而不是增加新的 block type。 +- 增加事件类型、增加必需字段、改变字段类型或语义、改变 envelope 或排序规则，必须提升 `schema_version`。 +- 版本升级必须同步更新中英文协议、producer、reader/replay 和 contract tests。 +- 未知顶层字段目前可被 reader 忽略，但不保证 deserialize/serialize 后保留；扩展应优先放在有明确所有权的 payload 或 `extension` 中。 + +Event Journal v1 是内部协议，并不承诺跨 nanoPyCodeAgent 大版本永远兼容。`schema_version` 的目的，是让不兼容变化被明确拒绝，而不是被静默误读。 + +## 实现位置速查 + +| 行为 | 位置 | +|---|---| +| 事件类型、payload 校验、content/usage schema | [`event_journal.py`](../../../src/nanopycodeagent/event_journal.py) | +| Journal Entry 编码、截断、权限、append、fsync、replay | [`event_journal.py`](../../../src/nanopycodeagent/event_journal.py) | +| run/user/model/tool 事件发射 | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| Anthropic block 到 provider-neutral content 的归一化 | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| Native Event 到现有 stdout 的文本投影 | [`agent.py`](../../../src/nanopycodeagent/agent.py) | +| envelope、顺序、权限、截断、partial tail 和 schema 测试 | [`test_event_journal.py`](../../../tests/test_event_journal.py) | +| 事件生命周期及精确 stdout 投影回归测试 | [`test_agent_events.py`](../../../tests/test_agent_events.py) | +| 原有 agent loop 与 stdout 行为测试 | [`test_agent.py`](../../../tests/test_agent.py) | diff --git a/docs/dev_notes/en/0.8.x.md b/docs/dev_notes/en/0.8.x.md index 3d26101..d23b580 100644 --- a/docs/dev_notes/en/0.8.x.md +++ b/docs/dev_notes/en/0.8.x.md @@ -172,3 +172,7 @@ The implementation order is: 5. **Implement a one-way ATIF projector.** Replay the Event Journal, mapping `user.message` to a user step and folding one model call plus its tool calls and results into an agent step. Map timestamps, tokens, cache, cost, terminal state, and `final_metrics`; only values with reliable sources and complete attribution enter standard fields, with other information under `extra`. At each checkpoint or run end, update the complete ATIF snapshot through a temporary file and atomic rename. 6. **Connect the CLI and Harbor.** Add an independent `--trajectory PATH`, preserve stdout's current text behavior, and reject `-` to avoid competing for stdout. The Harbor adapter only passes the log path, declares and reads ATIF, and then populates steps, tokens, and cost; it no longer converts a native trajectory. `--output-format` and `stream-json` remain for a later independent PR. 7. **Validate in layers.** Start with unit tests for event order, ID associations, failed tools, pending and resolved cost, Journal replay, and the ATIF schema. Then test stdout independence, atomic CLI writes, and interruption recovery. Finally, use Harbor contract tests and one real trial to confirm that the trajectory is collected and that step, token, and cost statistics are populated. + +This change completes steps 1–3 first. `event_journal.py` defines schema-version-1 Native Events and Journal Entries and creates an independent JSONL file for every Agent Run. `run.started` records the producer name and the nanoPyCodeAgent version resolved from package metadata; `seq` strictly increases within a run, `recorded_at` is UTC, and directory and file permissions are restricted to `0700` and `0600`, respectively. The writer appends complete lines, while replay ignores the final incomplete record left by an interrupted process and rejects records that move backward. Large strings are truncated only in the persisted copy with their JSON Pointer and original/retained lengths recorded; the core's Native Event remains unchanged. + +The agent loop now emits events at user, model, tool, and run boundaries, with an additional `model.output_delta` event preserving the existing streaming display. `model.completed` retains provider-neutral complete content and tool calls, the actual model, stop reason, token and cache usage, provider response ID, and `X-Generation-Id`; tool events retain inputs, results, error state, and precise duration. Model text and tool call/result stdout are now entirely projected from the same facts, and exact-output regression tests for a successful reply, a failed tool, and an interrupted stream show that the existing output is unchanged. Journals live under `~/.nanoPyCodeAgent/journals/` and are sensitive internal reconstruction data containing prompts, repository content, and tool results—not the public `--trajectory` format. Steps 4–7 are not implemented in this change. diff --git a/docs/dev_notes/zh-CN/0.8.x.md b/docs/dev_notes/zh-CN/0.8.x.md index f3bcab4..a6f3fbb 100644 --- a/docs/dev_notes/zh-CN/0.8.x.md +++ b/docs/dev_notes/zh-CN/0.8.x.md @@ -172,3 +172,7 @@ journal writer 接收它以后，补上持久化所需的身份、顺序和记 5. **实现单向 ATIF projector。** 重放 Event Journal，将 `user.message` 映射为 user step，将一次模型调用及其 tool calls/results 折叠成 agent step，并映射 timestamp、token、cache、cost、终态和 `final_metrics`；只有来源可靠、归因完整的值才进入标准字段，其余信息放入 `extra`。每次 checkpoint 或 run 结束时通过临时文件加原子 rename 更新完整 ATIF 快照。 6. **接通 CLI 与 Harbor。** 增加独立的 `--trajectory PATH`，保持 stdout 当前的文本行为，并拒绝 `-` 与 stdout 争用；Harbor adapter 只负责传入日志路径、声明并读取 ATIF，再回填 steps、tokens 和 cost，不再做 native trajectory 转换。`--output-format` 与 `stream-json` 留给后续独立 PR。 7. **分层验收。** 先用单元测试覆盖事件顺序、ID 关联、异常工具、pending/resolved cost、Journal 重放和 ATIF schema；再验证 CLI 的 stdout 独立性、原子写入和中断恢复；最后用 Harbor 契约测试与一次真实 trial 确认 trajectory 被采集，步骤与 token/cost 统计能够回填。 + +本次先完成了第 1～3 步。`event_journal.py` 定义了 schema version 1 的 Native Event 与 Journal Entry，并为每次 Agent Run 生成独立 JSONL；`run.started` 记录 producer name 与从包元数据解析的 nanoPyCodeAgent version，`seq` 在 run 内严格递增，`recorded_at` 使用 UTC，目录与文件权限分别收紧为 `0700` 和 `0600`。writer 以追加方式写入完整行，replay 会忽略进程中断留下的最后一条不完整记录，同时拒绝倒序记录；大字符串只在持久化副本中截断，并保留 JSON Pointer 与原始/保留长度，core 中的 Native Event 不受影响。 + +agent loop 现在在 user、model、tool 与 run 边界发出事件，另外用 `model.output_delta` 保留原有流式显示。`model.completed` 保存 provider-neutral 的完整 content/tool calls、实际 model、stop reason、token/cache usage、provider response ID 与 `X-Generation-Id`；工具事件保存输入、结果、错误状态和精确 duration。模型文本与工具调用/结果的 stdout 全部改由同一组事件投影，针对成功回复、失败工具和流中断的精确输出回归测试证明现有输出没有变化。Journal 位于 `~/.nanoPyCodeAgent/journals/`，是包含提示词、仓库内容和工具结果的敏感内部重建数据，不是 `--trajectory` 的公开格式；本轮没有实现计划第 4～7 步。 diff --git a/docs/research/en/agent_events_to_atif_examples.md b/docs/research/en/agent_events_to_atif_examples.md index 452e443..e5894a6 100644 --- a/docs/research/en/agent_events_to_atif_examples.md +++ b/docs/research/en/agent_events_to_atif_examples.md @@ -679,8 +679,7 @@ Native Events do not need to allocate persistence order themselves. After accept "result": "fn main() { println!(\"hello\"); }", "is_error": false, "duration_ms": 330, - "source_timestamp": null, - "timestamp_source": "receiver" + "source_timestamp": null } } ``` @@ -690,7 +689,7 @@ The semantic constraints should be: - `schema_version` describes the internal Journal Entry and Native Event contract, not the ATIF schema version. - `seq` is allocated by the journal writer, strictly increases, and is authoritative ordering within a run. - `recorded_at` is the UTC wall-clock time when nano accepted or recorded the event, uses RFC 3339/ISO 8601, and is required on every Journal Entry. -- `source_timestamp` belongs to the Native Event payload and is set only when a Source Record provides reliable source time. When absent, keep it `null`; never present `recorded_at` as source occurrence time. +- `source_timestamp` belongs to the Native Event payload. nano core timestamps the fact boundary; an adapter copies it only when a Source Record provides reliable source time, otherwise leaving it `null`. Never present `recorded_at` as source occurrence time. - Precise duration uses `duration_ms` or paired start/end events rather than relying only on subtraction between two wall-clock values. - ATIF `step.timestamp` prefers reliable `source_timestamp`, otherwise falls back to `recorded_at`, and records the choice in `extra.timestamp_source`. diff --git a/docs/research/zh-CN/agent_events_to_atif_examples.md b/docs/research/zh-CN/agent_events_to_atif_examples.md index 079ae08..129c4ed 100644 --- a/docs/research/zh-CN/agent_events_to_atif_examples.md +++ b/docs/research/zh-CN/agent_events_to_atif_examples.md @@ -679,8 +679,7 @@ Native Event 不必自行分配持久化顺序。journal writer 接受 Native Ev "result": "fn main() { println!(\"hello\"); }", "is_error": false, "duration_ms": 330, - "source_timestamp": null, - "timestamp_source": "receiver" + "source_timestamp": null } } ``` @@ -690,7 +689,7 @@ Native Event 不必自行分配持久化顺序。journal writer 接受 Native Ev - `schema_version` 描述内部 Journal Entry/Native Event 契约，不是 ATIF schema version； - `seq` 由 journal writer 分配，严格递增，是 run 内排序的权威； - `recorded_at` 是 nano 接受/记录事件时的 UTC wall-clock，使用 RFC 3339/ISO 8601，并且每条 Journal Entry 都必须有； -- `source_timestamp` 属于 Native Event payload，仅在 Source Record 提供可信原始时间时填写；缺失时保持 `null`，不能把 `recorded_at` 冒充成源端发生时间； +- `source_timestamp` 属于 Native Event payload；nano core 在事实边界采时，adapter 仅在 Source Record 提供可信原始时间时复制，无法确定时保持 `null`，不能把 `recorded_at` 冒充成源端发生时间； - 精确耗时使用 `duration_ms` 或成对 start/end 事件，不用两个 wall-clock 相减作为唯一依据； - ATIF `step.timestamp` 优先采用可信 `source_timestamp`，否则回退到 `recorded_at`，并在 `extra.timestamp_source` 记录来源。 diff --git a/src/nanopycodeagent/agent.py b/src/nanopycodeagent/agent.py index d7d772d..7273473 100644 --- a/src/nanopycodeagent/agent.py +++ b/src/nanopycodeagent/agent.py @@ -21,6 +21,8 @@ import os import sys +import time +import uuid from importlib.metadata import PackageNotFoundError, version try: @@ -39,6 +41,14 @@ from .bash_tool import BASH_TOOL, run_bash from .edit_tool import EDIT_TOOL, edit_preview, run_edit +from .event_journal import ( + EventEmitter, + EventJournal, + JsonObject, + JsonValue, + NativeEvent, + utc_now, +) from .read_tool import READ_TOOL, run_read from .settings import load_settings_env from .terminal import Spinner, print_tool_output, print_tool_use @@ -51,7 +61,7 @@ # How many model replies one headless task may spend before the run stops on # its own. The interactive loop needs no such cap — a human watching the -# transcript can interrupt a model that keeps retrying the same command — +# visible output can interrupt a model that keeps retrying the same command — # but an unattended run would keep paying for that loop until the API # refuses it. DEFAULT_MAX_TURNS = 50 @@ -88,6 +98,120 @@ TOOLS = [READ_TOOL, WRITE_TOOL, EDIT_TOOL, BASH_TOOL] +def _json_value(value: object) -> JsonValue: + """Convert an SDK value into the provider-neutral event representation.""" + if value is None or isinstance(value, bool | int | float | str): + return value + if isinstance(value, dict): + return {str(key): _json_value(item) for key, item in value.items()} + if isinstance(value, list | tuple): + return [_json_value(item) for item in value] + model_dump = getattr(value, "model_dump", None) + if callable(model_dump): + return _json_value(model_dump(mode="json", exclude_none=True)) + attributes = getattr(value, "__dict__", None) + if isinstance(attributes, dict): + return { + key: _json_value(item) + for key, item in attributes.items() + if not key.startswith("_") + } + raise TypeError(f"cannot represent {type(value).__name__} as event JSON") + + +def _native_content_blocks(value: object) -> list[JsonValue]: + """Normalize Anthropic response blocks into provider-neutral content.""" + source_blocks = _json_value(value) + if not isinstance(source_blocks, list): + raise TypeError("model response content must be a list") + content: list[JsonValue] = [] + for source_block in source_blocks: + if not isinstance(source_block, dict): + raise TypeError("model response content blocks must be objects") + source_type = source_block.get("type") + if source_type == "text": + content.append({"type": "text", "text": source_block.get("text", "")}) + elif source_type == "tool_use": + content.append( + { + "type": "tool_call", + "tool_call_id": source_block.get("id"), + "tool_name": source_block.get("name"), + "input": source_block.get("input"), + } + ) + else: + # Preserve an unfamiliar block without declaring its SDK shape to + # be part of the core schema. A future transport can normalize a + # corresponding provider block into the same content vocabulary. + content.append( + { + "type": "extension", + "namespace": "anthropic", + "source_type": source_type, + "value": source_block, + } + ) + return content + + +def _response_header(stream: object, name: str) -> str | None: + response = getattr(stream, "response", None) + headers = getattr(response, "headers", None) + if headers is None: + return None + value = headers.get(name) + if value is None: + value = headers.get(name.title()) + return str(value) if value is not None else None + + +class _TextOutputProjector: + """Preserve the existing text Run Output as a Native Event projection.""" + + def __init__(self, reply_prefix: str) -> None: + self._reply_prefix = reply_prefix + self._model_calls_with_text: set[str] = set() + + def __call__(self, event: NativeEvent) -> None: + if event.type == "model.output_delta": + model_call_id = str(event.payload["model_call_id"]) + if model_call_id not in self._model_calls_with_text: + if self._reply_prefix: + print(self._reply_prefix, end="", flush=True) + self._model_calls_with_text.add(model_call_id) + print(str(event.payload["delta"]), end="", flush=True) + elif event.type == "model.completed": + model_call_id = str(event.payload["model_call_id"]) + if model_call_id in self._model_calls_with_text: + print() + elif event.type == "tool.started": + tool_name = str(event.payload["tool_name"]) + arguments = event.payload["input"] + if not isinstance(arguments, dict): + raise TypeError("tool input event payload must be an object") + if tool_name == "read": + print_tool_use(f"[read] {arguments['path']}") + elif tool_name == "write": + content = str(arguments["content"]) + print_tool_use( + f"[write] {arguments['path']}\n{content_preview(content)}" + ) + elif tool_name == "edit": + old_text = str(arguments["old_text"]) + new_text = str(arguments["new_text"]) + print_tool_use( + f"[edit] {arguments['path']}\n" + f"{edit_preview(old_text, new_text)}" + ) + else: + print_tool_use(f"[bash]$ {arguments['command']}") + elif event.type == "tool.completed": + result = event.payload["result"] + if isinstance(result, str): + print_tool_output(result) + + def _package_version() -> str: """Return the installed package version. @@ -102,39 +226,80 @@ def _package_version() -> str: return "unknown" -def _run_one_tool(block: ToolUseBlock) -> ToolResultBlockParam: - """Execute one ``tool_use`` block, echoing the call and its output.""" - if block.name == "read": - path = block.input["path"] - print_tool_use(f"[read] {path}") - output, is_error = run_read( - path, - offset=block.input.get("offset", 1), - limit=block.input.get("limit"), - ) - elif block.name == "write": - path = block.input["path"] - content = block.input["content"] - # The echo folds the content: the terminal shows where the write - # goes and how it starts, not hundreds of lines. - print_tool_use(f"[write] {path}\n{content_preview(content)}") - output, is_error = run_write(path, content) - elif block.name == "edit": - path = block.input["path"] - old_text = block.input["old_text"] - new_text = block.input["new_text"] - # The echo folds both sides into a small -/+ diff: the terminal - # shows what is being swapped, not the whole strings again. - print_tool_use(f"[edit] {path}\n{edit_preview(old_text, new_text)}") - output, is_error = run_edit( - path, old_text, new_text, replace_all=block.input.get("replace_all", False) +def _run_one_tool( + block: ToolUseBlock, + emitter: EventEmitter, + model_call_id: str, +) -> ToolResultBlockParam: + """Execute one ``tool_use`` block and emit its runtime facts.""" + tool_input = _json_value(block.input) + if not isinstance(tool_input, dict): + raise TypeError("tool input must be an object") + emitter.emit( + "tool.started", + { + "model_call_id": model_call_id, + "tool_call_id": block.id, + "tool_name": block.name, + "input": tool_input, + "source_timestamp": utc_now(), + }, + ) + tool_started_ns = time.perf_counter_ns() + try: + if block.name == "read": + path = block.input["path"] + output, is_error = run_read( + path, + offset=block.input.get("offset", 1), + limit=block.input.get("limit"), + ) + elif block.name == "write": + path = block.input["path"] + content = block.input["content"] + output, is_error = run_write(path, content) + elif block.name == "edit": + path = block.input["path"] + old_text = block.input["old_text"] + new_text = block.input["new_text"] + output, is_error = run_edit( + path, + old_text, + new_text, + replace_all=block.input.get("replace_all", False), + ) + else: # bash — the only other tool offered + command = block.input["command"] + with Spinner("Running..."): + output, is_error = run_bash(command) + except BaseException as exc: + emitter.emit( + "tool.completed", + { + "model_call_id": model_call_id, + "tool_call_id": block.id, + "tool_name": block.name, + "result": None, + "is_error": True, + "error": {"type": type(exc).__name__, "message": str(exc)}, + "duration_ms": (time.perf_counter_ns() - tool_started_ns) + / 1_000_000, + "source_timestamp": utc_now(), + }, ) - else: # bash — the only other tool offered - command = block.input["command"] - print_tool_use(f"[bash]$ {command}") - with Spinner("Running..."): - output, is_error = run_bash(command) - print_tool_output(output) + raise + emitter.emit( + "tool.completed", + { + "model_call_id": model_call_id, + "tool_call_id": block.id, + "tool_name": block.name, + "result": output, + "is_error": is_error, + "duration_ms": (time.perf_counter_ns() - tool_started_ns) / 1_000_000, + "source_timestamp": utc_now(), + }, + ) return { "type": "tool_result", "tool_use_id": block.id, @@ -189,12 +354,90 @@ def _run_exchange( False when ``max_turns`` replies were spent while it was still calling them — the caller decides what an exhausted budget means. """ + run_id = f"run-{uuid.uuid4()}" + run_started_ns = time.perf_counter_ns() + projector = _TextOutputProjector(reply_prefix) + with EventJournal.create(run_id) as journal: + emitter = EventEmitter(journal, projector) + emitter.emit( + "run.started", + { + "mode": "headless" if max_turns is not None else "interactive", + "model": model, + "max_turns": max_turns, + "producer": { + "name": "nanoPyCodeAgent", + "version": _package_version(), + }, + "source_timestamp": utc_now(), + }, + ) + user_content = messages[-1]["content"] + emitter.emit( + "user.message", + { + "message_id": f"user-{uuid.uuid4()}", + "content": _json_value(user_content), + "source_timestamp": utc_now(), + }, + ) + try: + finished = _run_model_loop( + client, + model, + messages, + system, + emitter=emitter, + max_turns=max_turns, + ) + except BaseException as exc: + emitter.emit( + "run.failed", + { + "error_type": type(exc).__name__, + "message": str(exc), + "duration_ms": (time.perf_counter_ns() - run_started_ns) + / 1_000_000, + "source_timestamp": utc_now(), + }, + ) + raise + emitter.emit( + "run.completed", + { + "outcome": "completed" if finished else "max_turns_exhausted", + "duration_ms": (time.perf_counter_ns() - run_started_ns) / 1_000_000, + "source_timestamp": utc_now(), + }, + ) + return finished + + +def _run_model_loop( + client: anthropic.Anthropic, + model: str, + messages: list[MessageParam], + system: str, + *, + emitter: EventEmitter, + max_turns: int | None, +) -> bool: + """Run model replies and tool calls for an already-started Agent Run.""" turns = 0 while True: # A spinner marks the wait for the reply; the first streamed # token replaces it with the reply prefix. A tool-only reply # streams no text, so the prefix is skipped for it entirely. - replied = False + model_call_id = f"model-{uuid.uuid4()}" + emitter.emit( + "model.started", + { + "model_call_id": model_call_id, + "model": model, + "source_timestamp": utc_now(), + }, + ) + model_started_ns = time.perf_counter_ns() # Stream the reply so text shows up as it is generated, then grab # the accumulated message for the conversation history. with Spinner() as spinner, client.messages.stream( @@ -205,15 +448,43 @@ def _run_exchange( messages=messages, ) as stream: for text in stream.text_stream: - if not replied: - spinner.stop() - if reply_prefix: - print(reply_prefix, end="", flush=True) - replied = True - print(text, end="", flush=True) + spinner.stop() + emitter.emit( + "model.output_delta", + { + "model_call_id": model_call_id, + "delta": text, + "source_timestamp": utc_now(), + }, + ) message = stream.get_final_message() - if replied: - print() + generation_id = _response_header(stream, "x-generation-id") + model_completed_ns = time.perf_counter_ns() + + content = _native_content_blocks(message.content) + tool_calls = [ + item + for item in content + if isinstance(item, dict) and item.get("type") == "tool_call" + ] + provider_response_id = getattr(message, "id", None) + usage = _json_value(getattr(message, "usage", None)) + payload: JsonObject = { + "model_call_id": model_call_id, + "message_id": str(provider_response_id or model_call_id), + "content": content, + "tool_calls": tool_calls, + "model": str(getattr(message, "model", None) or model), + "stop_reason": getattr(message, "stop_reason", None), + "usage": usage, + "provider_response_id": ( + str(provider_response_id) if provider_response_id is not None else None + ), + "generation_id": generation_id, + "duration_ms": (model_completed_ns - model_started_ns) / 1_000_000, + "source_timestamp": utc_now(), + } + emitter.emit("model.completed", payload) turns += 1 messages.append({"role": "assistant", "content": message.content}) @@ -226,7 +497,9 @@ def _run_exchange( # Every tool_use block needs a matching tool_result in the next # user message, or the API rejects the request. results = [ - _run_one_tool(block) for block in message.content if block.type == "tool_use" + _run_one_tool(block, emitter, model_call_id) + for block in message.content + if block.type == "tool_use" ] messages.append({"role": "user", "content": results}) diff --git a/src/nanopycodeagent/event_journal.py b/src/nanopycodeagent/event_journal.py new file mode 100644 index 0000000..08947b6 --- /dev/null +++ b/src/nanopycodeagent/event_journal.py @@ -0,0 +1,662 @@ +"""Versioned runtime facts and their internal append-only Event Journal. + +Journal files contain prompts, model replies, tool inputs, and tool results. +They are sensitive internal reconstruction data, not public run output or an +ATIF trajectory. +""" + +from __future__ import annotations + +import json +import os +import re +import threading +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Callable, Mapping + +from . import settings + +SCHEMA_VERSION = 1 +DEFAULT_MAX_STRING_CHARS = 100_000 + +EVENT_TYPES = frozenset( + { + "run.started", + "user.message", + "model.started", + "model.output_delta", + "model.completed", + "tool.started", + "tool.completed", + "run.completed", + "run.failed", + } +) + +type JsonValue = ( + None | bool | int | float | str | list[JsonValue] | dict[str, JsonValue] +) +type JsonObject = dict[str, JsonValue] +type EventSubscriber = Callable[[NativeEvent], None] + +_NON_TRUNCATABLE_FIELDS = frozenset( + { + "error_type", + "generation_id", + "message_id", + "mode", + "model", + "model_call_id", + "outcome", + "producer", + "provider_response_id", + "source_timestamp", + "stop_reason", + "tool_call_id", + "tool_name", + } +) + +_REQUIRED_PAYLOAD_FIELDS = { + "run.started": frozenset( + {"mode", "model", "max_turns", "producer", "source_timestamp"} + ), + "user.message": frozenset({"message_id", "content", "source_timestamp"}), + "model.started": frozenset({"model_call_id", "model", "source_timestamp"}), + "model.output_delta": frozenset( + {"model_call_id", "delta", "source_timestamp"} + ), + "model.completed": frozenset( + { + "model_call_id", + "message_id", + "content", + "tool_calls", + "model", + "stop_reason", + "usage", + "provider_response_id", + "generation_id", + "duration_ms", + "source_timestamp", + } + ), + "tool.started": frozenset( + {"tool_call_id", "tool_name", "input", "source_timestamp"} + ), + "tool.completed": frozenset( + { + "tool_call_id", + "tool_name", + "result", + "is_error", + "duration_ms", + "source_timestamp", + } + ), + "run.completed": frozenset({"outcome", "duration_ms", "source_timestamp"}), + "run.failed": frozenset( + {"error_type", "message", "duration_ms", "source_timestamp"} + ), +} + + +def utc_now() -> str: + """Return the current UTC time in the journal's RFC 3339 format.""" + return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def _validate_rfc3339_utc(value: str, field: str) -> None: + if re.fullmatch( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z", + value, + ) is None: + raise ValueError(f"{field} must be RFC 3339 UTC") + try: + datetime.fromisoformat(value.removesuffix("Z") + "+00:00") + except ValueError as exc: + raise ValueError(f"{field} must be RFC 3339 UTC") from exc + + +def _validate_recorded_at(value: str) -> None: + _validate_rfc3339_utc(value, "Journal Entry recorded_at") + + +def _require_string(payload: JsonObject, field: str, event_type: str) -> str: + value = payload[field] + if not isinstance(value, str) or not value: + raise ValueError(f"{event_type}.{field} must be a non-empty string") + return value + + +def _validate_tool_call(value: JsonValue, field: str) -> JsonObject: + if not isinstance(value, dict): + raise ValueError(f"{field} items must be objects") + if value.get("type") != "tool_call": + raise ValueError(f"{field} items must have type tool_call") + for name in ("tool_call_id", "tool_name"): + item = value.get(name) + if not isinstance(item, str) or not item: + raise ValueError(f"{field}.{name} must be a non-empty string") + if not isinstance(value.get("input"), dict): + raise ValueError(f"{field}.input must be an object") + return value + + +def _validate_model_content(content: list[JsonValue]) -> list[JsonObject]: + tool_calls: list[JsonObject] = [] + for block in content: + if not isinstance(block, dict): + raise ValueError("model.completed.content items must be objects") + block_type = block.get("type") + if block_type == "text": + if not isinstance(block.get("text"), str): + raise ValueError("model.completed.content.text must be a string") + elif block_type == "tool_call": + tool_calls.append( + _validate_tool_call(block, "model.completed.content") + ) + elif block_type == "extension": + namespace = block.get("namespace") + if not isinstance(namespace, str) or not namespace: + raise ValueError( + "model.completed.content.extension namespace must be a string" + ) + source_type = block.get("source_type") + if source_type is not None and not isinstance(source_type, str): + raise ValueError( + "model.completed.content.extension source_type is invalid" + ) + if "value" not in block: + raise ValueError( + "model.completed.content.extension value is required" + ) + else: + raise ValueError(f"unsupported model content type: {block_type}") + return tool_calls + + +def _validate_usage(usage: JsonObject) -> None: + for field in ("input_tokens", "output_tokens"): + if field not in usage: + raise ValueError(f"model.completed.usage.{field} is required") + for field in ( + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + ): + if field not in usage: + continue + value = usage[field] + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError( + f"model.completed.usage.{field} must be non-negative" + ) + + +def _validate_json_value(value: object, field: str) -> None: + if value is None or isinstance(value, bool | int | float | str): + return + if isinstance(value, list): + for index, item in enumerate(value): + _validate_json_value(item, f"{field}[{index}]") + return + if isinstance(value, dict): + for key, item in value.items(): + if not isinstance(key, str): + raise ValueError(f"{field} object keys must be strings") + _validate_json_value(item, f"{field}.{key}") + return + raise ValueError(f"{field} must use only JSON data types") + + +def _validate_tool_error(error: JsonValue) -> None: + if not isinstance(error, dict): + raise ValueError("tool.completed.error is required when result is null") + error_type = error.get("type") + if not isinstance(error_type, str) or not error_type: + raise ValueError("tool.completed.error.type must be a non-empty string") + if not isinstance(error.get("message"), str): + raise ValueError("tool.completed.error.message must be a string") + + +def _validate_native_payload(event_type: str, payload: JsonObject) -> None: + required = _REQUIRED_PAYLOAD_FIELDS[event_type] + missing = sorted(required.difference(payload)) + if missing: + raise ValueError( + f"{event_type} missing required fields: {', '.join(missing)}" + ) + if "timestamp_source" in payload: + raise ValueError(f"{event_type}.timestamp_source is not supported") + + source_timestamp = payload["source_timestamp"] + if source_timestamp is not None: + if not isinstance(source_timestamp, str): + raise ValueError(f"{event_type}.source_timestamp must be RFC 3339 UTC") + _validate_rfc3339_utc( + source_timestamp, + f"{event_type}.source_timestamp", + ) + + for field in ("message_id", "model_call_id", "tool_call_id", "tool_name"): + if field in payload: + _require_string(payload, field, event_type) + + if "duration_ms" in payload: + duration = payload["duration_ms"] + if ( + not isinstance(duration, int | float) + or isinstance(duration, bool) + or duration < 0 + ): + raise ValueError(f"{event_type}.duration_ms must be non-negative") + + if event_type == "run.started": + if payload["mode"] not in {"interactive", "headless"}: + raise ValueError("run.started.mode must be interactive or headless") + _require_string(payload, "model", event_type) + producer = payload["producer"] + if not isinstance(producer, dict): + raise ValueError("run.started.producer must be an object") + for field in ("name", "version"): + value = producer.get(field) + if not isinstance(value, str) or not value: + raise ValueError( + f"run.started.producer.{field} must be a non-empty string" + ) + max_turns = payload["max_turns"] + if max_turns is not None and ( + not isinstance(max_turns, int) + or isinstance(max_turns, bool) + or max_turns < 1 + ): + raise ValueError("run.started.max_turns must be positive or null") + elif event_type == "model.started": + _require_string(payload, "model", event_type) + elif event_type == "model.output_delta": + if not isinstance(payload["delta"], str): + raise ValueError("model.output_delta.delta must be a string") + elif event_type == "model.completed": + _require_string(payload, "model", event_type) + if not isinstance(payload["content"], list): + raise ValueError("model.completed.content must be a list") + if not isinstance(payload["tool_calls"], list): + raise ValueError("model.completed.tool_calls must be a list") + content_tool_calls = _validate_model_content(payload["content"]) + tool_calls = [ + _validate_tool_call(item, "model.completed.tool_calls") + for item in payload["tool_calls"] + ] + if tool_calls != content_tool_calls: + raise ValueError( + "model.completed.tool_calls must match content tool calls" + ) + if payload["stop_reason"] is not None and not isinstance( + payload["stop_reason"], str + ): + raise ValueError("model.completed.stop_reason must be a string or null") + if payload["usage"] is not None and not isinstance(payload["usage"], dict): + raise ValueError("model.completed.usage must be an object or null") + if isinstance(payload["usage"], dict): + _validate_usage(payload["usage"]) + for field in ("provider_response_id", "generation_id"): + value = payload[field] + if value is not None and (not isinstance(value, str) or not value): + raise ValueError(f"model.completed.{field} must be a string or null") + elif event_type == "tool.started": + if not isinstance(payload["input"], dict): + raise ValueError("tool.started.input must be an object") + elif event_type == "tool.completed": + result = payload["result"] + if result is not None and not isinstance(result, str): + raise ValueError("tool.completed.result must be a string or null") + if not isinstance(payload["is_error"], bool): + raise ValueError("tool.completed.is_error must be a boolean") + if result is None: + if payload["is_error"] is not True: + raise ValueError( + "tool.completed.is_error must be true when result is null" + ) + _validate_tool_error(payload.get("error")) + elif event_type == "run.completed": + if payload["outcome"] not in {"completed", "max_turns_exhausted"}: + raise ValueError("run.completed.outcome is unsupported") + elif event_type == "run.failed": + _require_string(payload, "error_type", event_type) + if not isinstance(payload["message"], str): + raise ValueError("run.failed.message must be a string") + +@dataclass(frozen=True, slots=True) +class NativeEvent: + """One version-one runtime fact produced by the agent core.""" + + type: str + payload: JsonObject + + def __post_init__(self) -> None: + if self.type not in EVENT_TYPES: + raise ValueError(f"unsupported Native Event type: {self.type}") + if not isinstance(self.payload, dict): + raise ValueError("Native Event payload must be a JSON object") + _validate_json_value(self.payload, "Native Event payload") + try: + json.dumps(self.payload, allow_nan=False) + except (TypeError, ValueError) as exc: + raise ValueError("Native Event payload must be valid JSON") from exc + _validate_native_payload(self.type, self.payload) + + def to_dict(self) -> JsonObject: + """Return the version-one wire representation of this fact.""" + return {"type": self.type, "payload": self.payload} + + +class EventEmitter: + """Send each Native Event to durable history and live projectors.""" + + def __init__( + self, + journal: EventJournal, + *subscribers: EventSubscriber, + ) -> None: + self._journal = journal + self._subscribers = subscribers + + def emit(self, event_type: str, payload: JsonObject) -> NativeEvent: + """Create, persist, and project one runtime fact.""" + event = NativeEvent(event_type, payload) + self._journal.append(event) + for subscriber in self._subscribers: + subscriber(event) + return event + + +@dataclass(frozen=True, slots=True) +class JournalEntry: + """A Native Event with durable identity and ordering metadata.""" + + schema_version: int + run_id: str + seq: int + recorded_at: str + type: str + payload: JsonObject + truncation: JsonObject | None = None + + def to_dict(self) -> JsonObject: + """Return the JSONL record representation.""" + value: JsonObject = { + "schema_version": self.schema_version, + "run_id": self.run_id, + "seq": self.seq, + "recorded_at": self.recorded_at, + "type": self.type, + "payload": self.payload, + } + if self.truncation is not None: + value["truncation"] = self.truncation + return value + + @classmethod + def from_dict(cls, value: Mapping[str, object]) -> JournalEntry: + """Validate and rebuild one version-one Journal Entry.""" + schema_version = value.get("schema_version") + run_id = value.get("run_id") + seq = value.get("seq") + recorded_at = value.get("recorded_at") + event_type = value.get("type") + payload = value.get("payload") + truncation = value.get("truncation") + if ( + not isinstance(schema_version, int) + or isinstance(schema_version, bool) + or schema_version != SCHEMA_VERSION + ): + raise ValueError(f"unsupported Journal Entry schema: {schema_version}") + if not isinstance(run_id, str) or not run_id: + raise ValueError("Journal Entry run_id must be a non-empty string") + if not isinstance(seq, int) or isinstance(seq, bool) or seq < 1: + raise ValueError("Journal Entry seq must be a positive integer") + if not isinstance(recorded_at, str) or not recorded_at: + raise ValueError("Journal Entry recorded_at must be a non-empty string") + _validate_recorded_at(recorded_at) + if not isinstance(event_type, str) or not isinstance(payload, dict): + raise ValueError("Journal Entry must contain a Native Event") + if truncation is not None: + _validate_truncation(truncation) + event = NativeEvent(event_type, payload) + return cls( + schema_version=schema_version, + run_id=run_id, + seq=seq, + recorded_at=recorded_at, + type=event.type, + payload=event.payload, + truncation=truncation, + ) + + +def _json_pointer_part(value: str) -> str: + return value.replace("~", "~0").replace("/", "~1") + + +def _validate_truncation(value: object) -> None: + if not isinstance(value, dict): + raise ValueError("Journal Entry truncation must be an object") + _validate_json_value(value, "Journal Entry truncation") + fields = value.get("fields") + if not isinstance(fields, list) or not fields: + raise ValueError("Journal Entry truncation.fields must be a non-empty list") + for item in fields: + if not isinstance(item, dict): + raise ValueError("Journal Entry truncation fields must be objects") + path = item.get("path") + original_chars = item.get("original_chars") + retained_chars = item.get("retained_chars") + if not isinstance(path, str) or not path.startswith("/"): + raise ValueError( + "Journal Entry truncation field path must be a JSON Pointer" + ) + if ( + not isinstance(original_chars, int) + or isinstance(original_chars, bool) + or not isinstance(retained_chars, int) + or isinstance(retained_chars, bool) + or retained_chars < 0 + or original_chars <= retained_chars + ): + raise ValueError("Journal Entry truncation character counts are invalid") + + +def _is_non_truncatable_field(path: str, key: str) -> bool: + if key in _NON_TRUNCATABLE_FIELDS: + return True + return ( + key == "type" + and re.fullmatch(r"/(?:content|tool_calls)/\d+", path) is not None + ) + + +def _truncate_strings( + value: JsonValue, + *, + path: str, + limit: int, + fields: list[JsonObject], +) -> JsonValue: + if isinstance(value, str): + if len(value) <= limit: + return value + fields.append( + { + "path": path, + "original_chars": len(value), + "retained_chars": limit, + } + ) + return value[:limit] + if isinstance(value, list): + return [ + _truncate_strings( + item, + path=f"{path}/{index}", + limit=limit, + fields=fields, + ) + for index, item in enumerate(value) + ] + if isinstance(value, dict): + return { + key: ( + item + if _is_non_truncatable_field(path, key) + else _truncate_strings( + item, + path=f"{path}/{_json_pointer_part(key)}", + limit=limit, + fields=fields, + ) + ) + for key, item in value.items() + } + return value + + +class EventJournal: + """Append Journal Entries for one Agent Run to a sensitive JSONL file.""" + + def __init__( + self, + *, + path: Path, + run_id: str, + descriptor: int, + clock: Callable[[], str], + max_string_chars: int, + ) -> None: + self.path = path + self.run_id = run_id + self._descriptor = descriptor + self._clock = clock + self._max_string_chars = max_string_chars + self._next_seq = 1 + self._lock = threading.Lock() + + @classmethod + def create( + cls, + run_id: str, + *, + directory: Path | None = None, + clock: Callable[[], str] = utc_now, + max_string_chars: int = DEFAULT_MAX_STRING_CHARS, + ) -> EventJournal: + """Create a new journal file for ``run_id``.""" + if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", run_id) is None: + raise ValueError("run_id must be a safe filename component") + if max_string_chars < 1: + raise ValueError("max_string_chars must be positive") + if directory is None: + config_root = settings.SETTINGS_PATH.parent + config_root.mkdir(mode=0o700, parents=True, exist_ok=True) + config_root.chmod(0o700) + directory = config_root / "journals" + directory.mkdir(mode=0o700, parents=True, exist_ok=True) + directory.chmod(0o700) + path = directory / f"{run_id}.jsonl" + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_APPEND + flags |= getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags, 0o600) + os.fchmod(descriptor, 0o600) + return cls( + path=path, + run_id=run_id, + descriptor=descriptor, + clock=clock, + max_string_chars=max_string_chars, + ) + + def append(self, event: NativeEvent) -> JournalEntry: + """Wrap and append one runtime fact.""" + with self._lock: + if self._descriptor < 0: + raise ValueError("cannot append to a closed Event Journal") + truncated_fields: list[JsonObject] = [] + payload = _truncate_strings( + event.payload, + path="", + limit=self._max_string_chars, + fields=truncated_fields, + ) + assert isinstance(payload, dict) + recorded_at = self._clock() + _validate_recorded_at(recorded_at) + entry = JournalEntry( + schema_version=SCHEMA_VERSION, + run_id=self.run_id, + seq=self._next_seq, + recorded_at=recorded_at, + type=event.type, + payload=payload, + truncation={"fields": truncated_fields} if truncated_fields else None, + ) + encoded = json.dumps( + entry.to_dict(), ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + b"\n" + remaining = memoryview(encoded) + while remaining: + written = os.write(self._descriptor, remaining) + if written == 0: + raise OSError("could not append to Event Journal") + remaining = remaining[written:] + self._next_seq += 1 + return entry + + @staticmethod + def replay(path: Path) -> list[JournalEntry]: + """Read and validate the complete Journal Entries in ``path``.""" + entries: list[JournalEntry] = [] + with path.open("rb") as journal_file: + for raw_line in journal_file: + # The writer always terminates complete entries with a newline. + # A process killed during its final write can leave one partial + # tail; all entries before it remain independently replayable. + if not raw_line.endswith(b"\n"): + break + try: + value = json.loads(raw_line) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + line_number = len(entries) + 1 + raise ValueError( + f"invalid Journal Entry at line {line_number}" + ) from exc + if not isinstance(value, dict): + raise ValueError( + f"invalid Journal Entry at line {len(entries) + 1}" + ) + entry = JournalEntry.from_dict(value) + if entries and entry.run_id != entries[0].run_id: + raise ValueError("Event Journal contains more than one run_id") + if entries and entry.seq <= entries[-1].seq: + raise ValueError( + "Event Journal entries must have strictly increasing seq" + ) + entries.append(entry) + return entries + + def close(self) -> None: + """Close the underlying journal file.""" + with self._lock: + if self._descriptor >= 0: + os.fsync(self._descriptor) + os.close(self._descriptor) + self._descriptor = -1 + + def __enter__(self) -> EventJournal: + return self + + def __exit__(self, *exc_info: object) -> None: + self.close() diff --git a/tests/helpers.py b/tests/helpers.py index 5e6bade..cae014d 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -57,9 +57,22 @@ class FakeStream: tools. """ - def __init__(self, content, stop_reason="end_turn"): + def __init__( + self, + content, + stop_reason="end_turn", + *, + message_id="msg-test", + model="test-model", + usage=None, + response_headers=None, + ): self._content = content self._stop_reason = stop_reason + self._message_id = message_id + self._model = model + self._usage = usage + self.response = SimpleNamespace(headers=response_headers or {}) def __enter__(self): return self @@ -78,7 +91,13 @@ def _gen(): return _gen() def get_final_message(self): - return SimpleNamespace(content=self._content, stop_reason=self._stop_reason) + return SimpleNamespace( + id=self._message_id, + content=self._content, + model=self._model, + stop_reason=self._stop_reason, + usage=self._usage, + ) class FakeMessages: diff --git a/tests/test_agent_events.py b/tests/test_agent_events.py new file mode 100644 index 0000000..750fcdb --- /dev/null +++ b/tests/test_agent_events.py @@ -0,0 +1,261 @@ +"""End-to-end event and text-projection tests for the agent loop.""" + +from importlib.metadata import version +from types import SimpleNamespace + +import httpx +import pytest + +from nanopycodeagent import agent, settings +from nanopycodeagent.event_journal import EventJournal + +from helpers import ( + FakeClient, + FakeMessages, + FakeStream, + patch_client, + read_tool_use_block, + text_block, +) + + +def _only_journal_path(): + paths = list((settings.SETTINGS_PATH.parent / "journals").glob("*.jsonl")) + assert len(paths) == 1 + return paths[0] + + +def test_headless_model_reply_is_journaled_without_changing_stdout( + monkeypatch, capsys +): + usage = SimpleNamespace( + input_tokens=12, + output_tokens=3, + cache_read_input_tokens=5, + cache_creation_input_tokens=2, + ) + reply = FakeStream( + [text_block("done")], + message_id="msg-provider-1", + model="actual-model", + usage=usage, + response_headers={"x-generation-id": "gen-1"}, + ) + patch_client(monkeypatch, FakeClient(FakeMessages([reply]))) + + assert agent.run_headless("fix it") == 0 + + captured = capsys.readouterr() + assert captured.out == "done\n" + + entries = EventJournal.replay(_only_journal_path()) + assert [entry.type for entry in entries] == [ + "run.started", + "user.message", + "model.started", + "model.output_delta", + "model.completed", + "run.completed", + ] + assert [entry.seq for entry in entries] == [1, 2, 3, 4, 5, 6] + assert all("timestamp_source" not in entry.payload for entry in entries) + + started_run = entries[0].payload + assert started_run["producer"] == { + "name": "nanoPyCodeAgent", + "version": version("nanoPyCodeAgent"), + } + + user = entries[1].payload + assert user["content"] == "fix it" + assert user["message_id"] + + started = entries[2].payload + completed = entries[4].payload + assert completed["model_call_id"] == started["model_call_id"] + assert completed["message_id"] == "msg-provider-1" + assert completed["provider_response_id"] == "msg-provider-1" + assert completed["generation_id"] == "gen-1" + assert completed["model"] == "actual-model" + assert completed["stop_reason"] == "end_turn" + assert completed["content"] == [{"type": "text", "text": "done"}] + assert completed["tool_calls"] == [] + assert completed["usage"] == { + "input_tokens": 12, + "output_tokens": 3, + "cache_read_input_tokens": 5, + "cache_creation_input_tokens": 2, + } + assert completed["duration_ms"] >= 0 + assert completed["source_timestamp"].endswith("Z") + assert entries[-1].payload["outcome"] == "completed" + + +def test_model_duration_ends_before_local_response_normalization(monkeypatch): + reply = FakeStream([text_block("done")], message_id="msg-provider-1") + patch_client(monkeypatch, FakeClient(FakeMessages([reply]))) + clock_calls = [] + + def monotonic_ns(): + clock_calls.append(None) + return len(clock_calls) * 1_000_000 + + original_normalize = agent._native_content_blocks + + def normalize_after_model_timing(value): + assert len(clock_calls) == 3 + return original_normalize(value) + + monkeypatch.setattr(agent.time, "perf_counter_ns", monotonic_ns) + monkeypatch.setattr(agent, "_native_content_blocks", normalize_after_model_timing) + + assert agent.run_headless("fix it") == 0 + + entries = EventJournal.replay(_only_journal_path()) + completed = next(entry for entry in entries if entry.type == "model.completed") + assert completed.payload["duration_ms"] == 1 + + +def test_failed_tool_events_project_the_existing_tool_output( + monkeypatch, capsys, tmp_path +): + missing = tmp_path / "missing.txt" + tool_reply = FakeStream( + [read_tool_use_block("tool-1", path=str(missing))], + stop_reason="tool_use", + message_id="msg-tools", + ) + final_reply = FakeStream([text_block("recovered")], message_id="msg-final") + patch_client( + monkeypatch, + FakeClient(FakeMessages([tool_reply, final_reply])), + ) + + assert agent.run_headless("read the file") == 0 + + captured = capsys.readouterr() + assert captured.out == ( + f"[read] {missing}\n[file not found: {missing}]\nrecovered\n" + ) + + entries = EventJournal.replay(_only_journal_path()) + assert [entry.type for entry in entries] == [ + "run.started", + "user.message", + "model.started", + "model.completed", + "tool.started", + "tool.completed", + "model.started", + "model.output_delta", + "model.completed", + "run.completed", + ] + first_model_call_id = entries[2].payload["model_call_id"] + first_model_completed = entries[3].payload + started = entries[4].payload + completed = entries[5].payload + expected_tool_call = { + "type": "tool_call", + "tool_call_id": "tool-1", + "tool_name": "read", + "input": {"path": str(missing)}, + } + assert first_model_completed["content"] == [expected_tool_call] + assert first_model_completed["tool_calls"] == [expected_tool_call] + assert started["model_call_id"] == first_model_call_id + assert started["tool_call_id"] == "tool-1" + assert started["tool_name"] == "read" + assert started["input"] == {"path": str(missing)} + assert started["source_timestamp"].endswith("Z") + assert "timestamp_source" not in started + assert completed["model_call_id"] == first_model_call_id + assert completed["tool_call_id"] == "tool-1" + assert completed["tool_name"] == "read" + assert completed["result"] == f"[file not found: {missing}]" + assert completed["is_error"] is True + assert completed["duration_ms"] >= 0 + assert "timestamp_source" not in completed + + +def test_interrupted_model_stream_preserves_partial_stdout_and_records_failure( + monkeypatch, capsys +): + class DisconnectingStream(FakeStream): + @property + def text_stream(self): + def _chunks(): + yield "partial" + raise httpx.ReadError( + "peer disconnected", + request=httpx.Request("POST", "https://example.test/messages"), + ) + + return _chunks() + + patch_client( + monkeypatch, + FakeClient(FakeMessages([DisconnectingStream([])])), + ) + + assert agent.run_headless("keep going") == 1 + + captured = capsys.readouterr() + assert captured.out == "partial" + assert "API error: peer disconnected" in captured.err + + entries = EventJournal.replay(_only_journal_path()) + assert [entry.type for entry in entries] == [ + "run.started", + "user.message", + "model.started", + "model.output_delta", + "run.failed", + ] + failure = entries[-1].payload + assert failure["error_type"] == "ReadError" + assert failure["message"] == "peer disconnected" + assert failure["duration_ms"] >= 0 + assert "timestamp_source" not in failure + + +def test_unexpected_tool_exception_is_completed_before_the_run_fails( + monkeypatch, capsys, tmp_path +): + target = tmp_path / "notes.txt" + tool_reply = FakeStream( + [read_tool_use_block("tool-raises", path=str(target))], + stop_reason="tool_use", + ) + patch_client(monkeypatch, FakeClient(FakeMessages([tool_reply]))) + + def raise_from_read(*args, **kwargs): + raise RuntimeError("disk disappeared") + + monkeypatch.setattr(agent, "run_read", raise_from_read) + + with pytest.raises(RuntimeError, match="disk disappeared"): + agent.run_headless("read notes") + + captured = capsys.readouterr() + assert captured.out == f"[read] {target}\n" + + entries = EventJournal.replay(_only_journal_path()) + assert [entry.type for entry in entries] == [ + "run.started", + "user.message", + "model.started", + "model.completed", + "tool.started", + "tool.completed", + "run.failed", + ] + tool_failure = entries[-2].payload + assert tool_failure["tool_call_id"] == "tool-raises" + assert tool_failure["result"] is None + assert tool_failure["is_error"] is True + assert tool_failure["error"] == { + "type": "RuntimeError", + "message": "disk disappeared", + } + assert tool_failure["duration_ms"] >= 0 diff --git a/tests/test_event_journal.py b/tests/test_event_journal.py new file mode 100644 index 0000000..affbd7f --- /dev/null +++ b/tests/test_event_journal.py @@ -0,0 +1,518 @@ +"""Behavioral tests for runtime events and the append-only Event Journal.""" + +import json +import stat + +import pytest + +from nanopycodeagent import settings +from nanopycodeagent.event_journal import EventJournal, JournalEntry, NativeEvent + + +def _run_started_event(): + return NativeEvent( + "run.started", + { + "mode": "headless", + "model": "test-model", + "max_turns": 50, + "producer": {"name": "nanoPyCodeAgent", "version": "0.8.0"}, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + +def test_journal_entry_wraps_the_native_event_with_ordering_metadata(tmp_path): + event = NativeEvent( + "tool.completed", + { + "tool_call_id": "call-read-1", + "tool_name": "read", + "result": "file contents", + "is_error": False, + "duration_ms": 330, + "source_timestamp": None, + }, + ) + + with EventJournal.create( + "run-123", + directory=tmp_path, + clock=lambda: "2026-08-23T08:00:01.420Z", + ) as journal: + entry = journal.append(event) + + assert event.to_dict() == { + "type": "tool.completed", + "payload": { + "tool_call_id": "call-read-1", + "tool_name": "read", + "result": "file contents", + "is_error": False, + "duration_ms": 330, + "source_timestamp": None, + }, + } + assert entry.to_dict() == { + "schema_version": 1, + "run_id": "run-123", + "seq": 1, + "recorded_at": "2026-08-23T08:00:01.420Z", + "type": "tool.completed", + "payload": event.payload, + } + + with pytest.raises(ValueError, match="unsupported Native Event type"): + NativeEvent("trajectory.step", {}) + + +def test_jsonl_journal_replays_entries_in_append_order(tmp_path): + timestamps = iter( + ["2026-08-23T08:00:00.000Z", "2026-08-23T08:00:00.001Z"] + ) + with EventJournal.create( + "run-order", + directory=tmp_path, + clock=lambda: next(timestamps), + ) as journal: + path = journal.path + journal.append( + NativeEvent( + "user.message", + { + "message_id": "user-1", + "content": "hello", + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + ) + journal.append( + NativeEvent( + "run.completed", + { + "outcome": "completed", + "duration_ms": 12.5, + "source_timestamp": "2026-08-23T08:00:00.001Z", + }, + ) + ) + + entries = EventJournal.replay(path) + + assert [entry.seq for entry in entries] == [1, 2] + assert [entry.type for entry in entries] == ["user.message", "run.completed"] + assert entries[0].payload["content"] == "hello" + assert path.read_bytes().count(b"\n") == 2 + + +def test_journal_storage_is_restricted_to_the_current_user(tmp_path): + directory = tmp_path / "journals" + directory.mkdir(mode=0o755) + directory.chmod(0o755) + + with EventJournal.create( + "run-sensitive", + directory=directory, + clock=lambda: "2026-08-23T08:00:00.000Z", + ) as journal: + path = journal.path + + assert stat.S_IMODE(directory.stat().st_mode) == 0o700 + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_default_journal_storage_restricts_the_configuration_root( + monkeypatch, tmp_path +): + config_root = tmp_path / "config" + config_root.mkdir(mode=0o755) + config_root.chmod(0o755) + monkeypatch.setattr(settings, "SETTINGS_PATH", config_root / "settings.json") + + with EventJournal.create( + "run-default-storage", + clock=lambda: "2026-08-23T08:00:00.000Z", + ) as journal: + path = journal.path + + assert stat.S_IMODE(config_root.stat().st_mode) == 0o700 + assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700 + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_large_strings_are_truncated_only_in_the_persisted_entry(tmp_path): + event = NativeEvent( + "tool.completed", + { + "tool_call_id": "tool-1", + "tool_name": "read", + "result": "abcdefghij", + "is_error": False, + "duration_ms": 10, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + with EventJournal.create( + "run-large-output", + directory=tmp_path, + clock=lambda: "2026-08-23T08:00:00.000Z", + max_string_chars=4, + ) as journal: + entry = journal.append(event) + path = journal.path + + assert event.payload["result"] == "abcdefghij" + assert entry.payload["result"] == "abcd" + assert entry.to_dict()["truncation"] == { + "fields": [ + { + "path": "/result", + "original_chars": 10, + "retained_chars": 4, + } + ] + } + assert EventJournal.replay(path) == [entry] + + +def test_producer_identity_is_not_truncated(tmp_path): + with EventJournal.create( + "run-producer-version", + directory=tmp_path, + clock=lambda: "2026-08-23T08:00:00.000Z", + max_string_chars=4, + ) as journal: + entry = journal.append(_run_started_event()) + + assert entry.payload["producer"] == { + "name": "nanoPyCodeAgent", + "version": "0.8.0", + } + assert entry.truncation is None + + +def test_model_content_types_remain_replayable_at_the_smallest_limit(tmp_path): + tool_call = { + "type": "tool_call", + "tool_call_id": "tool-1", + "tool_name": "read", + "input": {"path": "README.md"}, + } + event = NativeEvent( + "model.completed", + { + "model_call_id": "model-1", + "message_id": "message-1", + "content": [{"type": "text", "text": "hello"}, tool_call], + "tool_calls": [tool_call], + "model": "test-model", + "stop_reason": "tool_use", + "usage": None, + "provider_response_id": None, + "generation_id": None, + "duration_ms": 25, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + with EventJournal.create( + "run-content-types", + directory=tmp_path, + clock=lambda: "2026-08-23T08:00:00.000Z", + max_string_chars=1, + ) as journal: + entry = journal.append(event) + path = journal.path + + assert [block["type"] for block in entry.payload["content"]] == [ + "text", + "tool_call", + ] + assert entry.payload["tool_calls"][0]["type"] == "tool_call" + assert EventJournal.replay(path) == [entry] + + +def test_replay_ignores_only_a_trailing_partial_entry(tmp_path): + with EventJournal.create( + "run-interrupted", + directory=tmp_path, + clock=lambda: "2026-08-23T08:00:00.000Z", + ) as journal: + expected = journal.append(_run_started_event()) + path = journal.path + + with path.open("ab") as journal_file: + journal_file.write(b'{"schema_version":1,"run_id":"run-interrupted"') + + assert EventJournal.replay(path) == [expected] + + +def test_run_id_cannot_escape_the_journal_directory(tmp_path): + with pytest.raises(ValueError, match="safe filename component"): + EventJournal.create( + "../escape", + directory=tmp_path, + clock=lambda: "2026-08-23T08:00:00.000Z", + ) + + assert list(tmp_path.parent.glob("escape.jsonl")) == [] + + +def test_recorded_at_must_be_an_rfc3339_utc_timestamp(tmp_path): + with EventJournal.create( + "run-invalid-time", + directory=tmp_path, + clock=lambda: "2026-08-23 08:00:00", + ) as journal: + with pytest.raises(ValueError, match="recorded_at must be RFC 3339 UTC"): + journal.append(_run_started_event()) + + +def test_replay_rejects_non_increasing_sequence_numbers(tmp_path): + with EventJournal.create( + "run-bad-order", + directory=tmp_path, + clock=lambda: "2026-08-23T08:00:00.000Z", + ) as journal: + first = journal.append(_run_started_event()) + path = journal.path + + duplicate = first.to_dict() + with path.open("a", encoding="utf-8") as journal_file: + journal_file.write(json.dumps(duplicate) + "\n") + + with pytest.raises(ValueError, match="strictly increasing seq"): + EventJournal.replay(path) + + +def test_native_event_contract_rejects_missing_fields_and_untrusted_measurements(): + with pytest.raises(ValueError, match="run.started missing required fields: producer"): + NativeEvent( + "run.started", + { + "mode": "headless", + "model": "test-model", + "max_turns": 50, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + with pytest.raises(ValueError, match="model.completed missing required fields"): + NativeEvent("model.completed", {}) + + with pytest.raises(ValueError, match="duration_ms must be non-negative"): + NativeEvent( + "run.completed", + { + "outcome": "completed", + "duration_ms": -1, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + with pytest.raises(ValueError, match="source_timestamp must be RFC 3339 UTC"): + NativeEvent( + "run.failed", + { + "error_type": "RuntimeError", + "message": "failed", + "duration_ms": 1, + "source_timestamp": "yesterday", + }, + ) + + +@pytest.mark.parametrize("content", [("not", "json"), {1: "not-json"}]) +def test_native_event_contract_rejects_non_json_values(content): + with pytest.raises(ValueError, match="JSON data types|keys must be strings"): + NativeEvent( + "user.message", + { + "message_id": "user-1", + "content": content, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + +def test_journal_entry_rejects_boolean_schema_version(): + with pytest.raises(ValueError, match="unsupported Journal Entry schema"): + JournalEntry.from_dict( + { + "schema_version": True, + "run_id": "run-1", + "seq": 1, + "recorded_at": "2026-08-23T08:00:00.000Z", + "type": "user.message", + "payload": { + "message_id": "user-1", + "content": "hello", + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + } + ) + + +@pytest.mark.parametrize( + "truncation", + [ + {"fields": "not-a-list"}, + {"fields": [{1: "not-json"}]}, + {"fields": [{}]}, + { + "fields": [ + {"path": "/content", "original_chars": 1, "retained_chars": 1} + ] + }, + ], +) +def test_journal_entry_rejects_invalid_truncation_metadata(truncation): + with pytest.raises(ValueError, match="Journal Entry truncation"): + JournalEntry.from_dict( + { + "schema_version": 1, + "run_id": "run-1", + "seq": 1, + "recorded_at": "2026-08-23T08:00:00.000Z", + "type": "user.message", + "payload": { + "message_id": "user-1", + "content": "h", + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + "truncation": truncation, + } + ) + + +@pytest.mark.parametrize( + ("is_error", "error"), + [ + (False, {"type": "RuntimeError", "message": "failed"}), + (True, {}), + (True, {"type": "", "message": "failed"}), + (True, {"type": "RuntimeError", "message": None}), + ], +) +def test_tool_completed_rejects_invalid_null_result(is_error, error): + with pytest.raises(ValueError, match="tool.completed"): + NativeEvent( + "tool.completed", + { + "tool_call_id": "tool-1", + "tool_name": "read", + "result": None, + "is_error": is_error, + "error": error, + "duration_ms": 10, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + +def test_native_event_contract_rejects_removed_timestamp_source(): + with pytest.raises(ValueError, match="timestamp_source is not supported"): + NativeEvent( + "run.completed", + { + "outcome": "completed", + "duration_ms": 10, + "source_timestamp": "2026-08-23T08:00:00.000Z", + "timestamp_source": "core", + }, + ) + + +def test_replay_rejects_removed_timestamp_source(tmp_path): + path = tmp_path / "removed-timestamp-source.jsonl" + path.write_text( + json.dumps( + { + "schema_version": 1, + "run_id": "run-removed-timestamp-source", + "seq": 1, + "recorded_at": "2026-08-23T08:00:00.001Z", + "type": "run.completed", + "payload": { + "outcome": "completed", + "duration_ms": 10, + "source_timestamp": "2026-08-23T08:00:00.000Z", + "timestamp_source": "core", + }, + } + ) + + "\n", + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="timestamp_source is not supported"): + EventJournal.replay(path) + + +@pytest.mark.parametrize( + "producer", + [ + None, + {}, + {"name": "nanoPyCodeAgent"}, + {"name": "", "version": "0.8.0"}, + {"name": "nanoPyCodeAgent", "version": ""}, + ], +) +def test_run_started_requires_valid_producer_identity(producer): + with pytest.raises(ValueError, match="run.started.producer"): + NativeEvent( + "run.started", + { + "mode": "headless", + "model": "test-model", + "max_turns": 50, + "producer": producer, + "source_timestamp": "2026-08-23T08:00:00.000Z", + }, + ) + + +def test_model_completed_contract_validates_nested_content_and_usage(): + payload = { + "model_call_id": "model-1", + "message_id": "message-1", + "content": [], + "tool_calls": [], + "model": "test-model", + "stop_reason": "end_turn", + "usage": {"input_tokens": 10, "output_tokens": 2}, + "provider_response_id": "message-1", + "generation_id": None, + "duration_ms": 25, + "source_timestamp": "2026-08-23T08:00:00.000Z", + } + + with pytest.raises(ValueError, match="content items must be objects"): + NativeEvent("model.completed", payload | {"content": [42]}) + + with pytest.raises(ValueError, match="tool_call_id must be a non-empty string"): + invalid_call = {"type": "tool_call", "tool_name": "read", "input": {}} + NativeEvent( + "model.completed", + payload | {"content": [invalid_call], "tool_calls": [invalid_call]}, + ) + + with pytest.raises(ValueError, match="usage.input_tokens must be non-negative"): + NativeEvent( + "model.completed", + payload | {"usage": {"input_tokens": "10", "output_tokens": 2}}, + ) + + with pytest.raises(ValueError, match="tool_calls must match content tool calls"): + valid_call = { + "type": "tool_call", + "tool_call_id": "tool-1", + "tool_name": "read", + "input": {"path": "README.md"}, + } + NativeEvent( + "model.completed", + payload | {"content": [valid_call], "tool_calls": []}, + )