diff --git a/CHANGELOG.md b/CHANGELOG.md index af12934..f24f5a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **Breaking: conversation folders.** A `capture.agent_session` / + `agent_session_async` turn with a `conversation_id` is written to + `captures//conv_<16 hex>/cap_.json`, on disk and in an object + store. The hex is the start of `sha256(conversation_id)`. Standalone + captures stay at `captures//cap_.json`, and so does every + capture from the `@capture.agent` decorator, whose `conversation_id` is + fixed when it is applied (its envelope still carries the id). **Requires + evalshift CLI 1.6.0 or later, so upgrade the CLI first**: older CLIs do not + see the folders, and their `capture clean` can sweep toolset sidecars that + only folder captures use. If an evalshift CLI older than 1.6 is installed + in the same environment, the SDK writes flat and logs a warning (set + `EVALSHIFT_CAPTURE_LAYOUT=conversation` to override); an older CLI in a + *different* environment (e.g. CI pinned to 1.5) can't be detected, so + upgrade it first. Code that globs `captures//*.json` must also read + `captures//conv_*/*.json`. Set `EVALSHIFT_CAPTURE_LAYOUT=flat` (or + pass `layout="flat"` to `FileSink` / `ObjectStoreSink`) to keep the 0.6 + layout. +- **Sampling keeps conversations whole.** With `EVALSHIFT_SAMPLE_RATE` set, + a `capture.agent_session` (or `agent_session_async`) given a + `conversation_id` is sampled by a hash of that id, so every turn of a + conversation is captured or none is, in every process. The + `conversation_id` is what gets sampled: a user id passed as + `conversation_id` samples users, and a constant id in `agent_session` + keeps every run or none. The `@capture.agent` decorator (whose + `conversation_id` is fixed when it is applied), runs without an id, and + the LangChain handler still draw at random. +- **`max_captures` / `capture_ttl` count conversations.** Each suite + directory is capped by unit, a flat capture or a whole conversation + folder. `capture_ttl` first evicts every capture older than it; + `max_captures` then counts the captures the TTL leaves, newest unit first: + the newest conversations are kept whole while they fit, the one at the + boundary keeps its newest turns, and older ones are evicted whole. So one + newer capture evicts at most one older capture, not a whole live + conversation, and a conversation that never ends (a user id passed as + `conversation_id`) keeps its newest `max_captures` turns and stays bounded. + Folders left empty are removed. + Suites with only flat captures behave as before. Dedup is unchanged: its + hash already includes the conversation and turn. - The PyPI project page's Homepage link goes to instead of the GitHub repo, and a Documentation link to is added. diff --git a/DOCS.md b/DOCS.md index 5841c1b..f37d2eb 100644 --- a/DOCS.md +++ b/DOCS.md @@ -108,7 +108,7 @@ Run it with capture enabled: EVALSHIFT_CAPTURE=1 python agent.py ``` -One capture file appears at `.evalshift/captures/support_demo/cap_.json` (relative to the current working directory). Trimmed real output: +One capture file appears at `.evalshift/captures/support_demo/cap_.json` (relative to the current working directory). A `@capture.agent` decorator capture is always written there, flat; a turn of an `agent_session` with a `conversation_id` goes in its conversation's folder instead (see [Multi-turn conversations](#multi-turn-conversations)). Trimmed real output: ```json { @@ -195,6 +195,8 @@ call to @capture.agent-wrapped function │ 3. serialize: span tree → ordered trace events → capture envelope ▼ sink write ──── dedup check → base sink write (disk, memory, or object store) → GC (disk only) + (an agent_session turn with a conversation_id goes in conv_<16 hex>/; GC caps + the suite dir by unit: a capture or a whole conversation folder) ``` Span-to-event mapping: a **tool span expands into two events** — a `tool_call` at its start and a `tool_result` at its close. A `model_call` or `error` span becomes one event. Events get a dense `sequence_index` ordered by `(timestamp, monotonic op-order)`, so concurrent operations serialize deterministically. Span timing and parentage are stashed under each event's `metadata["evalshift"]` block. @@ -521,7 +523,9 @@ for turn_index, user_text in enumerate(user_turns): Notes: - The SDK does not return the written `capture_id` to the caller, so `parent_capture_id` can only be set if you manage capture ids yourself (or omit it — `conversation_id` + `turn_index` are enough to reconstruct order). -- When `conversation_id` is set, turn identity is folded into `input_hash`, so short repeated turns ("yes", "1pm") don't collapse under dedup. Standalone captures (`conversation_id=None`) hash exactly as before 1.1.0. +- When `conversation_id` is set, turn identity is folded into `input_hash`, so short repeated turns ("yes", "1pm") don't collapse under dedup. Standalone captures (`conversation_id=None`) hash exactly as before 1.1.0. Conversation folders (below) change nothing here: dedup never drops a distinct turn. +- **Where the turns go.** An `agent_session` / `agent_session_async` turn with a `conversation_id` is written to `captures//conv_<16 hex>/cap_.json`, where the hex is the start of `sha256(conversation_id)` (never the raw id); every other capture, including every `@capture.agent` decorator capture, is written to `captures//cap_.json`. The decorator's `conversation_id` is fixed when it is applied, so it names no conversation; its envelope still carries the id. The layout is the same on disk and in an object store, and reading the folders needs evalshift CLI 1.6.0 or later. If an evalshift CLI older than 1.6 is installed in the same environment, the SDK writes flat and logs a warning (set `EVALSHIFT_CAPTURE_LAYOUT=conversation` to override); an older CLI in a *different* environment (e.g. CI pinned to 1.5) can't be detected, so upgrade it first. `EVALSHIFT_CAPTURE_LAYOUT=flat` (or `layout="flat"` on `FileSink` / `ObjectStoreSink`) writes every capture flat, as before 0.7.0. +- **Sampling keeps a conversation whole; GC keeps the newest ones whole.** With a sample rate set, the decision is made once per agent invocation; per conversation when an `agent_session` is given a `conversation_id` (a hash of the id, so every turn is captured or none is, in every process). GC works per suite dir, by unit: a capture or a whole conversation folder (see [Keeping `captures/` bounded](#keeping-captures-bounded)). A conversation longer than `max_captures`, or at the boundary of the count, or older than `capture_ttl`, loses its oldest turns locally, and older conversations go whole, so pass the full messages list to each turn's model call. ### Error capture semantics @@ -544,7 +548,7 @@ configure( sink=MemorySink(), # where captures go (default: FileSink) sample_rate=0.25, # capture ~25% of runs (default: capture all) dedup=True, # collapse identical-input captures (default: True) - max_captures=200, # newest-N cap per suite dir (default: 200) + max_captures=200, # newest-N cap per suite dir, by unit (default: 200) capture_ttl=7 * 86400, # evict captures older than N seconds (default: off) require_model_call=True, # drop captures with no model_call span (default: False) ) @@ -560,10 +564,11 @@ configure( |---|---|---|---| | `EVALSHIFT_CAPTURE` | unset (off) | Master gate. Truthy values: `1`, `true`, `yes`, `on` (case-insensitive). Anything else is off. | **Live, on every call** — can be toggled mid-process | | `EVALSHIFT_DIR` | `.evalshift` | Capture root directory. Relative paths resolve against the **current working directory** (no repo-root walk). | **Live, at each write** | -| `EVALSHIFT_MAX_CAPTURES` | `200` | Keep only the newest N capture files per suite directory. | At config construction (import or `reset_config()`) | +| `EVALSHIFT_MAX_CAPTURES` | `200` | Keep only the newest N capture files per suite dir, by unit: a capture or a whole conversation folder. | At config construction (import or `reset_config()`) | | `EVALSHIFT_CAPTURE_TTL` | off | Evict capture files older than N seconds. | At config construction | | `EVALSHIFT_DEDUP` | `on` | Collapse captures with identical `(suite, input_hash)` within one process. | At config construction | -| `EVALSHIFT_SAMPLE_RATE` | off (capture all) | Capture only this fraction of runs (0.0–1.0), decided once per agent invocation. | At config construction | +| `EVALSHIFT_SAMPLE_RATE` | off (capture all) | Capture only this fraction of runs (0.0–1.0), decided once per agent invocation; per conversation when an `agent_session` is given a `conversation_id`. | At config construction | +| `EVALSHIFT_CAPTURE_LAYOUT` | unset (conversation folders) | `flat` (any case, stripped) writes every capture flat, as before 0.7.0; `conversation` writes conversation folders even when an evalshift CLI older than 1.6 is installed in the same environment. Unset or any other value means conversation folders, unless such an older CLI is installed alongside: then the SDK writes flat and logs one warning per process. A `layout=` argument on `FileSink` / `ObjectStoreSink` wins over it. | **Live, at each write** (the installed-CLI check runs once per process) | | `EVALSHIFT_CAPTURE_STORE` | unset (local disk) | Ship captures to an object store you own: `s3:///`, `gs:///` or `az:////`. Needs the provider's client library (`pip install boto3` / `google-cloud-storage` / `azure-storage-blob azure-identity`). An unusable value — library missing, malformed URI — is recorded here and raised as `SinkConfigurationError` by the first `capture.*` decorator, client wrapper, LangChain handler or `configure()` call while `EVALSHIFT_CAPTURE` is on; the message names the variable, the scheme and the fix, never the value. With capture off nothing happens. `EVALSHIFT_SINK`, its 0.5.0 name, still works when this is unset. | At config construction | - For the numeric knobs, `0`, `none`, `unlimited`, or `off` means "no cap / disabled". Disable @@ -581,8 +586,8 @@ configure( By default a long-running host cannot grow `captures/` without limit; three mechanisms compose: - **Dedup** (on by default): a per-process registry keyed on `(suite, input_hash)` drops repeat captures of the same input. The registry resets on process restart. -- **GC** (200 files per suite by default): after each successful disk write, the suite directory is capped to the newest `max_captures` files (ordered by filesystem mtime) and files older than `capture_ttl` seconds are evicted. GC never recurses into subdirectories and never raises. -- **Sampling** (off by default): with `sample_rate=0.1`, ~10% of agent invocations are captured; unsampled runs skip all bookkeeping entirely. A fault in the sampling draw fails open to *capturing*. +- **GC** (200 files per suite by default): after each successful disk write, GC works per suite dir, by unit: a capture or a whole conversation folder, timed by its newest file (filesystem mtime). First, files older than `capture_ttl` seconds are evicted one by one, so an idle conversation goes whole and a live one loses only its old turns. Then `max_captures` counts the captures the TTL leaves: units newest first (by their newest surviving file), turns within a conversation newest first, and the first `max_captures` are kept. So the newest conversations are kept whole while they fit, the one at the boundary keeps its newest turns, and older ones are evicted whole; one newer capture evicts at most one older capture, not a whole live conversation, and a conversation that never ends (a user id passed as `conversation_id`) keeps only its newest `max_captures` turns. A folder left empty is removed. A suite with only flat captures is capped exactly as before 0.7.0. GC enters only conversation folders, one level, never `toolsets/`, skips a folder or file it cannot read, and never raises. +- **Sampling** (off by default): with `sample_rate=0.1`, ~10% of agent invocations are captured, decided once per agent invocation; per conversation when an `agent_session` is given a `conversation_id`, so every turn of a conversation is captured or none is, in every process. The `conversation_id` is what gets sampled: a user id passed as `conversation_id` samples users, and a constant id in `agent_session` keeps every run or none. The `@capture.agent` decorator (whose `conversation_id` is fixed when it is applied), runs without an id and the LangChain handler draw at random. Unsampled runs skip all bookkeeping entirely. A fault in the sampling draw fails open to *capturing*. To restore unbounded write-everything behavior: @@ -590,7 +595,7 @@ To restore unbounded write-everything behavior: EVALSHIFT_MAX_CAPTURES=0 EVALSHIFT_DEDUP=off ``` -**With an object store** (`EVALSHIFT_CAPTURE_STORE`), dedup still applies but `max_captures` and `capture_ttl` do not — GC only ever runs after a real disk write. Bound the bucket with a lifecycle rule instead (e.g. expire objects under `/captures/` after 30 days). Dedup is per process, so several hosts writing the same input each store a capture; `evalshift capture sync` drops content duplicates when it promotes, so the bucket may be noisy while the suite is not. +**With an object store** (`EVALSHIFT_CAPTURE_STORE`), dedup still applies but `max_captures` and `capture_ttl` do not — GC only ever runs after a real disk write. Bound the bucket with a lifecycle rule instead (e.g. expire objects under `/captures/` after 30 days; the rule covers the conversation folders under it too). A lifecycle rule expires each object by its own age, so a live conversation's oldest turns can expire before its newest. Dedup is per process, so several hosts writing the same input each store a capture; `evalshift capture sync` drops content duplicates when it promotes, so the bucket may be noisy while the suite is not. **`require_model_call`** (off by default) is a persistence gate for eval-grade capture: a capture with no `model_call` event carries no scoreable ground truth, so hosts that only want promotable captures enable `configure(require_model_call=True)` and content-free captures are silently dropped (one debug log line). @@ -607,8 +612,9 @@ class Sink(Protocol): ### `FileSink` (default) -Writes `/captures//.json` as UTF-8 JSON. +Writes `/captures//.json` as UTF-8 JSON, or `/captures//conv_<16 hex>/.json` for an `agent_session` / `agent_session_async` turn with a `conversation_id` (the hex is the start of `sha256(conversation_id)`). Every `@capture.agent` decorator capture is written flat. +- **Layout:** `FileSink(layout="flat")` or `EVALSHIFT_CAPTURE_LAYOUT=flat` writes every capture flat, as before 0.7.0; `layout="conversation"` or `EVALSHIFT_CAPTURE_LAYOUT=conversation` writes conversation folders; the argument wins over the env var. With neither, the SDK writes conversation folders unless an evalshift CLI older than 1.6 is installed in the same environment, when it writes flat and logs one warning per process. Reading the folders needs evalshift CLI 1.6.0 or later. - **Base resolution, at write time:** constructor argument > `EVALSHIFT_DIR` env var > `.evalshift` relative to the current working directory. There is deliberately no repo-root walk — the SDK must not assume it runs inside a checkout (containers, Lambda). - Returns the **absolute** `Path` written, or `None` when the write degrades on a filesystem `OSError` (read-only mount, disk full, permission denied) — the capture is dropped, the host never crashes. - The `suite` name is sanitized into a single safe path segment (separators and `..` replaced), so a suite name can never escape `/captures/`. @@ -665,7 +671,7 @@ EVALSHIFT_CAPTURE=1 EVALSHIFT_CAPTURE_STORE=s3://acme-evals/support-agent python `` is optional. Credentials never go in the URI — a URI containing `@` or `?` is rejected, and the error does not echo it back. If `EVALSHIFT_CAPTURE_STORE` is invalid, its library is missing, or the store cannot be opened for any other reason, the SDK records the problem at import and raises it as `SinkConfigurationError` the first time your code asks it to capture — applying a `capture.*` decorator, wrapping a client, building the LangChain handler or calling `configure()` without a `sink` — while `EVALSHIFT_CAPTURE` is on. In a real agent that is process start. Bare `import evalshift` never raises, and with capture off nothing happens. Should the gate turn on only after those ran, captures are dropped with one `WARNING` rather than written to local disk: a store you asked for is never silently replaced. The message names the variable, the scheme and the package to install (`Run: pip install boto3`), never the value. -**The keys are the local layout.** Under your prefix the sink writes exactly what `FileSink` and `ToolsetSink` write to disk — `captures//cap_.json` and `toolsets/.json` — so the CLI can mirror the bucket into `.evalshift/` and read it unchanged (`evalshift capture sync` does this automatically once `captures.store` is set in `evalshift.yaml`; see the CLI docs). Until your CLI version supports `captures.store`, `aws s3 sync` / `gcloud storage rsync` / `azcopy sync` into a local `.evalshift/` followed by `evalshift capture sync` works the same way. +**The keys are the local layout.** Under your prefix the sink writes exactly what `FileSink` and `ToolsetSink` write to disk — `captures//cap_.json`, `captures//conv_<16 hex>/cap_.json` for an `agent_session` turn with a `conversation_id`, and `toolsets/.json` — so the CLI can mirror the bucket into `.evalshift/` and read it unchanged (`evalshift capture sync` does this automatically once `captures.store` is set in `evalshift.yaml`; see the CLI docs). Until your CLI version supports `captures.store`, `aws s3 sync` / `gcloud storage rsync` / `azcopy sync` into a local `.evalshift/` followed by `evalshift capture sync` works the same way. Reading the conversation folders needs evalshift CLI 1.6.0 or later; `ObjectStoreSink(store, layout="flat")` or `EVALSHIFT_CAPTURE_LAYOUT=flat` writes every capture key flat, as before 0.7.0. **Toolset sidecars go to the same store**, once per process per toolset. If a background sidecar upload fails, the sink forgets that toolset so the next model call offering it uploads the sidecar again; captures stamped in between point at a missing sidecar, and the CLI refuses to promote them. @@ -864,6 +870,7 @@ from pathlib import Path from evalshift import load_capture, load_envelope raw = Path(".evalshift/captures/support_demo/cap_abc.json").read_text() +# a conversation's turns: .evalshift/captures//conv_<16 hex>/cap_.json data = load_capture(raw) # dict, migrated to the current schema version env = load_envelope(raw) # typed CaptureEnvelope dataclass @@ -923,7 +930,7 @@ Top-level exports: `capture`, `record_model_call`, `configure`, `flush_captures` ) ``` -Decorator capturing one agent invocation per call. Works on `def` and `async def` (auto-detected). No-op unless the `EVALSHIFT_CAPTURE` gate is on. Agent input is derived by binding call arguments to the signature. Conversation kwargs are static per decoration — use `agent_session` for per-turn values. `tools` sets this session's toolset — inherited by any `record_model_call` / `capture.model_call` inside the decorated function that passes its own `tools=None`; a call's own value always overrides it (D-toolset). +Decorator capturing one agent invocation per call. Works on `def` and `async def` (auto-detected). No-op unless the `EVALSHIFT_CAPTURE` gate is on. Agent input is derived by binding call arguments to the signature. Conversation kwargs are static per decoration — use `agent_session` for per-turn values. Because a static `conversation_id` names no conversation, the decorator's captures are always written flat and sampled at random per run, even with one; the envelope still carries the id. `tools` sets this session's toolset — inherited by any `record_model_call` / `capture.model_call` inside the decorated function that passes its own `tools=None`; a call's own value always overrides it (D-toolset). ### `capture.agent_session` / `capture.agent_session_async` @@ -942,7 +949,7 @@ with capture.agent_session( ... ``` -Context-manager form of `capture.agent` for inline instrumentation; `agent_session_async` is the identical `async with` form. Yields the live `SpanTree`, or `None` when capture is off or the run wasn't sampled. **Always pass `agent_input`** — it feeds `input_hash`, the dedup key. The recommended primitive for multi-turn conversations (fresh `turn_index` per `with`). `tools` behaves exactly as on `capture.agent`. +Context-manager form of `capture.agent` for inline instrumentation; `agent_session_async` is the identical `async with` form. Yields the live `SpanTree`, or `None` when capture is off or the run wasn't sampled. **Always pass `agent_input`** — it feeds `input_hash`, the dedup key. The recommended primitive for multi-turn conversations (fresh `turn_index` per `with`). With a `conversation_id`, the turn is written to `captures//conv_<16 hex>/`, and a sample rate keeps or drops the conversation whole (see [Multi-turn conversations](#multi-turn-conversations)). `tools` behaves exactly as on `capture.agent`. ### `capture.model_call` @@ -1069,11 +1076,11 @@ class Sink(Protocol): ### `FileSink` ```python -FileSink(base: str | os.PathLike[str] | None = None) +FileSink(base: str | os.PathLike[str] | None = None, *, layout: Literal["conversation", "flat"] | None = None) FileSink.write(envelope) -> Path | None ``` -Writes `/captures//.json`. Base resolution at write time: constructor arg > `EVALSHIFT_DIR` > CWD-relative `.evalshift`. Returns the absolute path, or `None` on `OSError` (dropped, logged at debug). Suite segment sanitized against path traversal. +Writes `/captures//.json`, or `/captures//conv_<16 hex>/.json` for an `agent_session` turn with a `conversation_id` (`@capture.agent` decorator captures are always flat). `layout="flat"` or `EVALSHIFT_CAPTURE_LAYOUT=flat` writes every capture flat; the argument wins over the env var. Base resolution at write time: constructor arg > `EVALSHIFT_DIR` > CWD-relative `.evalshift`. Returns the absolute path, or `None` on `OSError` (dropped, logged at debug). Suite segment sanitized against path traversal. ### `MemorySink` @@ -1092,8 +1099,8 @@ In-memory buffer for read-only filesystems and tests. Thread-safe. from evalshift import ObjectStoreSink from evalshift.stores import ObjectStore, MemoryStore, parse_store_uri, open_store -ObjectStoreSink(store: ObjectStore, *, background: bool = True, queue_size: int = 1000, flush_timeout: float = 10.0) -ObjectStoreSink.write(envelope) -> None +ObjectStoreSink(store: ObjectStore, *, background: bool = True, queue_size: int = 1000, flush_timeout: float = 10.0, layout: Literal["conversation", "flat"] | None = None) +ObjectStoreSink.write(envelope) -> None # key: FileSink's path relative to , same layout rule ObjectStoreSink.write_toolset(normalized, fingerprint) -> str | None ObjectStoreSink.flush(timeout: float | None = None) -> bool ObjectStoreSink.store -> ObjectStore @@ -1180,7 +1187,7 @@ Return a drop-in proxy over a provider client instance that records one `model_c Work down this checklist: 1. **Gate off** — is `EVALSHIFT_CAPTURE` set to `1`/`true`/`yes`/`on`? Anything else (including unset) means off. -2. **Wrong directory** — the default `.evalshift` is relative to the process's *current working directory*, not the repo root. Set `EVALSHIFT_DIR` for a stable location. +2. **Wrong directory** — the default `.evalshift` is relative to the process's *current working directory*, not the repo root. Set `EVALSHIFT_DIR` for a stable location. An `agent_session` turn with a `conversation_id` is one level down, in `captures//conv_<16 hex>/` (or flat, with one `WARNING` in the log, when an evalshift CLI older than 1.6 is installed in the same environment). 3. **Dedup collapsed it** — same `(suite, input_hash)` already written this process. The classic trigger: `agent_session` without `agent_input=` (every session hashes identically). Disable with `EVALSHIFT_DEDUP=off` to test. 4. **Sampling skipped it** — is `sample_rate` / `EVALSHIFT_SAMPLE_RATE` set? 5. **`require_model_call` dropped it** — the gate is on and the run recorded no `model_call` event. diff --git a/README.md b/README.md index f458e0f..d115941 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,15 @@ own when the host's disk does not outlive it. The `evalshift` CLI reads those ca disk, or mirrors them down from the bucket; the layout is the interface, and the SDK and CLI never call each other. +A `capture.agent_session` / `agent_session_async` turn with a `conversation_id` is written to +`captures//conv_<16 hex>/cap_.json`, where the hex is the start of +`sha256(conversation_id)`; every other capture, including every `@capture.agent` decorator +capture, is written to `captures//cap_.json`. Reading the folders needs evalshift CLI +1.6.0 or later: if an older evalshift CLI is installed in the same environment, the SDK writes +flat and logs a warning (`EVALSHIFT_CAPTURE_LAYOUT=conversation` overrides it); an older CLI in a +different environment can't be detected, so upgrade it first. `EVALSHIFT_CAPTURE_LAYOUT=flat` (or +`layout="flat"` on `FileSink` / `ObjectStoreSink`) writes every capture flat, as before 0.7.0. + - **Distribution:** `evalshift-sdk` · **import name:** `evalshift` - **Runtime deps:** none (stdlib-only) - **Python:** >= 3.10 @@ -86,15 +95,18 @@ client = wrap_openai(OpenAI()) # OpenAI(base_url=...) c Capture writes one JSON file per sampled invocation, so `.evalshift/captures/` is kept bounded by default (no configuration needed): identical-input re-runs are de-duplicated, and each suite -directory is capped at the **200 newest** captures (oldest evicted). Tune it with env vars — no code -change required (precedence: an explicit `configure(...)` call > env var > built-in default): +directory is capped at the **200 newest** captures the TTL leaves, counted by unit: a capture or +a whole conversation folder. The newest conversations are kept whole while they fit, the one at +the boundary keeps its newest turns, and older ones are evicted whole. Tune it with env vars — no +code change required (precedence: an explicit `configure(...)` call > env var > built-in default): | Env var | Default | Meaning | | --- | --- | --- | -| `EVALSHIFT_MAX_CAPTURES` | `200` | Max captures kept per suite dir. `0` / `none` / `unlimited` = uncapped. | +| `EVALSHIFT_MAX_CAPTURES` | `200` | Max captures kept per suite dir, by unit: a capture or a whole conversation folder. `0` / `none` / `unlimited` = uncapped. | | `EVALSHIFT_DEDUP` | `on` | Collapse identical-input captures (per-process). `off` to disable. | | `EVALSHIFT_CAPTURE_TTL` | off | Evict captures older than N **seconds**. | -| `EVALSHIFT_SAMPLE_RATE` | off | Capture only this fraction of runs, e.g. `0.25`. | +| `EVALSHIFT_SAMPLE_RATE` | off | Capture only this fraction of runs, e.g. `0.25`; decided once per agent invocation; per conversation when an `agent_session` is given a `conversation_id`. | +| `EVALSHIFT_CAPTURE_LAYOUT` | conversation folders (flat if an evalshift CLI older than 1.6 is installed alongside) | `flat` writes every capture flat, as before 0.7.0; `conversation` writes conversation folders even beside an older CLI. | | `EVALSHIFT_DIR` | `.evalshift` | Capture root directory. | A malformed value falls back to the default (these knobs never crash capture). To restore fully unbounded diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 65b6365..90d791d 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -180,6 +180,9 @@ exactly as in Phases 0-5; `active_sink()` returns the bare sink with identity pr - **Sampling decides at agent entry**, not write time. `should_capture_now()` is read at the top of `_run_agent`/`_run_agent_async` and both `agent_session*` context managers; an unsampled run is a near-pure pass-through (no span tree, no redaction, no serialize). Decision made once per run. + *Amended by D-conv-layout:* an `agent_session` / `agent_session_async` given a + `conversation_id` draws by a hash of it, so the decision is per conversation, the same in every + process; the decorator, runs without an id and the LangChain handler still draw at random. - **Dedup + GC compose as a `HygieneSink`** returned by `active_sink()` only when a knob is set — the single write seam in `capture/api.py` is unchanged. Fail-open is **granular inside the wrapper**: the outer `fail_open("sink write")` would drop the whole capture, so the dedup check @@ -200,6 +203,11 @@ exactly as in Phases 0-5; `active_sink()` returns the bare sink with identity pr back, so one `os.scandir` stat pass (no JSON parse) gives both count-ordering and TTL-age cheaply, even at large `max_captures`. Limitation: tools that rewrite mtimes (`cp -p`, rsync, tar-extract) can perturb eviction order. `capture_ttl` is in **seconds**. + *Amended by D-conv-layout:* GC works per suite dir, by unit: a flat capture, or a whole + `conv_<16 hex>/` conversation folder timed by its newest file's mtime (one more `os.scandir` + per folder, still no JSON parse). The TTL goes first; the count then keeps the newest units + whole while they fit, the unit at the boundary keeps its newest turns, older units are evicted + whole, and a folder left empty is removed. - **Verified by:** `tests/test_hygiene_{sample,dedup,gc,sink}.py`, `tests/test_capture_sampling.py`, `tests/test_capture_hygiene_e2e.py`, and the hygiene fault matrix added to `tests/test_capture_failopen.py` + `tests/test_capture_async_failopen.py`. @@ -215,7 +223,10 @@ from the environment (no code change required): `sample_rate` stay `None` (off). 200 is larger than any realistic single golden suite, so normal use never loses recent data, but runaway accumulation is capped. Dedup-on only collapses *identical-input* re-runs (the same best-effort per-process key as D-6) — a free win against the - re-run-the-same-eval churn pattern. + re-run-the-same-eval churn pattern. *Amended by D-conv-layout:* the 200 are counted per suite + dir, by unit: a capture or a whole conversation folder, among the captures the TTL leaves. The + newest conversations are kept whole while they fit, the one at the boundary keeps its newest + turns, and older ones are evicted whole. - **Env vars set the default** (read at `_Config` construction and re-read on `reset_config()`): `EVALSHIFT_MAX_CAPTURES`, `EVALSHIFT_CAPTURE_TTL`, `EVALSHIFT_DEDUP`, `EVALSHIFT_SAMPLE_RATE`. For the numeric knobs, `0` / `none` / `unlimited` means **uncapped** (`None`); a malformed value @@ -366,6 +377,11 @@ policy in `docs/SCHEMA.md`. of, `/toolsets/`. Distinct toolsets are few and captures number in the thousands, so an evicted toolset would orphan every capture that still references it, not just the one just evicted; orphan sweeping (if ever needed) is a CLI-side concern, not this SDK's. + *Amended by D-conv-layout:* `evict` now enters `conv_<16 hex>` conversation folders, one level, + and never any other subdirectory (its docstring says so), and its caller still passes + `/captures//`, so `toolsets/` stays out of reach. An evalshift CLI older than 1.6.0 + does not see folder captures, so its `capture clean` can sweep sidecars that only they use: + upgrade the CLI first. - **No migration.** Schema 2.0.0 (D-8) already registers no edge from the 1.x major for exactly this reason — there is no honest value `tools_offered` can take for a capture written before per-call toolset capture existed. This decision does not reopen that question; it is the feature D-8's @@ -523,7 +539,10 @@ D-stores-c) or `configure(sink=...)`. - **Keys are the local layout** (`captures//cap_.json`, `toolsets/.json`), so the CLI mirrors a bucket into `.evalshift/` and reads it unchanged. The documented contract - moves from "disk is the only interface" to "the layout is the interface". + moves from "disk is the only interface" to "the layout is the interface". *Amended by + D-conv-layout:* an `agent_session` turn with a `conversation_id` is keyed + `captures//conv_<16 hex>/cap_.json`, the same relative key `FileSink` writes, + so the bucket stays a mirror of `.evalshift/`. - **One URI grammar** shared verbatim with the CLI (`stores/uri.py`): `s3://`, `gs://`, `az:////`; `@`/`?` rejected so credentials never live in config. Neither leaks into logs either: the `@`/`?` rejections do not echo the URI, and the @@ -578,6 +597,88 @@ stays a silent alias read only when the new name is unset or blank (no deprecati one day of exposure does not justify log noise), and `_store_env()` returns the variable name with the value so every `SinkConfigurationError` names the one the user set. +## Conversation folders (D-conv-layout) + +### D-conv-layout (2026-10-20) — a conversation's turns share one folder, sampling keeps it whole, and GC keeps the newest ones whole +**Problem.** Until 0.7.0 every capture was a flat `captures//cap_.json`, one file per +turn, so nothing in the layout said which files form a conversation; only the envelope's +`conversation_id` did, and reading it means parsing every file. Each piece that works per file +cut conversations apart. Sampling drew per run, so a 10% rate kept about one turn in ten of every +conversation. GC evicted the oldest files of a suite, so a long conversation lost its first turns +while its later ones stayed. And the CLI's per-suite budgets cut a bucket or a suite at a file +boundary, so they could take a conversation's later turns without its first. Each gap leaves the +CLI rebuilding a turn whose earlier turns are missing. + +**Decision.** An `agent_session` / `agent_session_async` turn with a `conversation_id` lives at +`captures//conv_<16 hex>/.json`, where the folder is `conv_` plus the first +16 hex digits of `sha256(str(conversation_id))` (`sinks/_layout.py`): lowercase, fixed length, +safe on every filesystem, and never the raw id, which can carry `:` or a user id. Every other +capture stays at `captures//.json`. Disk and store share the relative +layout, so a bucket stays a mirror of `.evalshift/`; toolset sidecars stay at `toolsets/.json`. +Captures from the `@capture.agent` decorator (sync and async) are always written flat, even with +a `conversation_id` (`api._finalize(flat=True)` wraps the write in `writing_flat()`): the id is +fixed when the function is decorated, so it names no conversation, and its runs would make one +never-ending folder that is the whole suite. Their envelope still carries the id. +`EVALSHIFT_CAPTURE_LAYOUT=flat` (any case, stripped) or `layout="flat"` on `FileSink` / +`ObjectStoreSink` writes every capture flat; `EVALSHIFT_CAPTURE_LAYOUT=conversation` or +`layout="conversation"` writes the folder layout; the argument wins over the env var. With +neither (unset, empty or a typo, so a typo can never stop captures), the SDK writes the folder +layout unless an evalshift CLI older than 1.6 is installed in the same environment (see +Compatibility). With a sample rate, +`agent_session` / `agent_session_async` given a `conversation_id` draw +`conversation_draw(conversation_id)` (`hygiene/sample.py`: 64 bits of +`sha256("evalshift-sample:" + id)`), so a conversation is kept or dropped whole in every process; +the decorator, runs without an id and the LangChain handler draw `random.random()`. +`max_captures` / `capture_ttl` work per suite dir, by unit: a flat capture, or a whole `conv_*` +folder timed by its newest file's mtime (`hygiene/gc.py`). The TTL goes first: it evicts every +file older than it, so an idle conversation goes whole and a live one loses only its old turns. +The count then runs on the files the TTL leaves: units ordered newest first by their newest +surviving file, the files within a unit newest first, and the first `max_captures` files in that +order are kept. So the newest units are kept whole while they fit, the unit at the boundary keeps +its newest turns that fit, and every older unit is evicted whole. A folder or file GC cannot read +(another uid, or gone mid-scan) is skipped, which only lowers the count. A folder left empty is +removed, and a suite with only flat files behaves exactly as before (the newest `max_captures`). +Dedup is unchanged: `input_hash` already folds in `conversation_id` and `turn_index` +(`trace/serialize.py`), so it never drops a distinct turn. +*Why not one mutable file per conversation:* object stores cannot append, so every turn would +rewrite the whole object; turns of one conversation land on different processes (behind a load +balancer, after a restart), whose read-modify-write cycles on the one file would race and lose +turns; and rewriting the full history on every turn makes a conversation's cost grow with the +square of its length. One immutable file per turn keeps every write a single create, and the +folder gives readers the grouping. + +**Trade-offs.** The newest unit keeps only its newest `max_captures` files, which is what keeps a +never-ending conversation (an `agent_session` given a user id as its `conversation_id`) bounded. +So a conversation longer than `max_captures`, or at the boundary of the count, or older than +`capture_ttl`, loses its oldest turns locally, and older conversations go whole; the CLI then +reports later rebuilt turns as missing earlier turns. That is the price of a bound; apps should +pass the full messages list. Keeping the boundary conversation's newest turns, rather than +evicting it whole, is what keeps one newer write from evicting more than one older capture (a +conversation at the cap plus one newer flat capture keeps its newest `max_captures - 1` turns). +The decorator's static id keeps drawing at random, since +hashing it would capture every run or none, forever, and keeps writing flat, since its folder +would hold the whole suite. And GC, or the CLI's +`capture clean`, can remove a folder it has just emptied in the instant another process writes a +new turn into it: that one write fails with an `OSError`, `FileSink` drops the capture and +returns `None`, fail-open, and nothing is raised into the host. + +**Compatibility.** The CLI ships first: evalshift CLI 1.6.0 reads both layouts (`/*.json`, +then `/conv_<16 hex>/*.json`, one level, never recursive), and SDK 0.7.0 requires +evalshift CLI 1.6.0 or later. An older CLI does not see the folders, and its `capture clean` can +sweep toolset sidecars that only folder captures use, so the CLI must be upgraded first. CLI 1.5.x +declares `evalshift-sdk>=0.4.0` with no upper bound, so installing it can pull SDK 0.7.0 in +without the user choosing it. So if an evalshift CLI older than 1.6 is installed in the same +environment (`importlib.metadata.version("evalshift")`, checked once per process, its leading +`major.minor`), the SDK writes flat and logs a warning (set `EVALSHIFT_CAPTURE_LAYOUT=conversation` +to override); an older CLI in a *different* environment (e.g. CI pinned to 1.5) can't be +detected, so upgrade it first. A missing CLI or an unreadable version means no guard. Code that +globs `captures//*.json` must also read `captures//conv_*/*.json`. +`EVALSHIFT_CAPTURE_LAYOUT=flat` is the escape hatch: a production rollback lever and a bridge +for tooling that globs the flat layout, with no code change. The envelope and `schema_version` +are unchanged. Amends D-6, D-6a, D-toolset's GC note and D-stores in place. Verified by +`tests/test_conversation_layout.py`, `tests/test_hygiene_gc.py`, `tests/test_hygiene_sample.py`, +`tests/test_capture_sampling.py` and `tests/test_hygiene_sink.py`. + ## Open follow-ups (not blocking v1) - ~~**D1-followup:** unify packaging so CLI + SDK co-install cleanly~~ — resolved 2026-09-09 in the CLI-depends-on-SDK form; see D-pkg. diff --git a/docs/SCHEMA.md b/docs/SCHEMA.md index a3a2652..5feea9b 100644 --- a/docs/SCHEMA.md +++ b/docs/SCHEMA.md @@ -51,7 +51,7 @@ replay: | Field | Type | Default | Meaning | |----------------------|---------------|---------|---------| -| `conversation_id` | `str \| None` | `None` | Shared identifier for every capture belonging to the same conversation. `None` for a standalone (single-turn) capture. | +| `conversation_id` | `str \| None` | `None` | Shared identifier for every capture belonging to the same conversation. `None` for a standalone (single-turn) capture. Since SDK 0.7.0 it also picks the file's folder: an `agent_session` / `agent_session_async` turn with one is written to `captures//conv_<16 hex>/cap_.json` (the hex is the start of `sha256(conversation_id)`); see [below](#where-a-conversations-turns-are-written). | | `turn_index` | `int \| None` | `None` | This capture's position in the conversation (0-based recommended; the SDK does not enforce an ordering). | | `parent_capture_id` | `str \| None` | `None` | The `capture_id` of the immediately preceding turn's capture, if any — lets a reader reconstruct turn order without relying on file mtimes. | @@ -83,6 +83,22 @@ multi-turn conversation (a different `turn_index` per call), use `capture.agent_ is no stateful `capture.conversation()` helper — turn identity is the caller's responsibility to track and pass in. +### Where a conversation's turns are written + +The folder a capture lands in is derived from the envelope, not stored in it, so it needed no +schema change (the envelope keys and `schema_version` are as above). Since SDK 0.7.0, a +`capture.agent_session` / `agent_session_async` turn with a `conversation_id` is written to +`captures//conv_<16 hex>/cap_.json`, where the folder is `conv_` plus the first 16 +lowercase hex digits of `sha256(str(conversation_id))`: fixed length, safe on every filesystem, +and never the raw id. Every other capture, including every `@capture.agent` decorator capture +(its static `conversation_id` names no conversation), stays at `captures//cap_.json`; +the decorator's envelope still carries the id. The layout is the same on disk and in an object +store. Reading the folders needs evalshift CLI 1.6.0 or later: if an older evalshift CLI is +installed in the same environment, the SDK writes flat and logs a warning +(`EVALSHIFT_CAPTURE_LAYOUT=conversation` overrides it). `EVALSHIFT_CAPTURE_LAYOUT=flat` (or +`layout="flat"` on `FileSink` / `ObjectStoreSink`) writes every capture flat, as before 0.7.0. +Rationale: `docs/DECISIONS.md` D-conv-layout. + ### The messages-list convention for `model_call.input` `capture.model_call(input=...)` was already untyped (`input: Any`) before 1.1.0, so no signature @@ -119,6 +135,10 @@ different turns of a conversation frequently share short, repeated user text — means two turns with identical `agent_input` text hash differently, so dedup no longer collapses them. +Conversation folders (SDK 0.7.0) leave dedup unchanged: it still keys on `(suite, input_hash)`, +and the hash already folds in `conversation_id` and `turn_index`, so dedup never drops a distinct +turn. + ## 2.0.0: per-call toolset fields, and no migration from 1.x Schema `2.0.0` added two optional fields to `model_call` events so the CLI can tell "this agent diff --git a/examples/support_agent/README.md b/examples/support_agent/README.md index 1b95802..5c31576 100644 --- a/examples/support_agent/README.md +++ b/examples/support_agent/README.md @@ -23,7 +23,9 @@ cd evalshift-sdk/examples/support_agent export EVALSHIFT_DIR="$PWD/.evalshift" # 1. capture (this checkout's SDK). Writes .evalshift/captures/support_demo/cap_*.json plus the -# content-addressed toolset sidecar .evalshift/toolsets/.json. +# content-addressed toolset sidecar .evalshift/toolsets/.json. The agent uses the +# @capture.agent decorator, whose captures are always flat; an agent_session turn with a +# conversation_id would go to .evalshift/captures/support_demo/conv_<16 hex>/ instead. EVALSHIFT_CAPTURE=1 uv run --project ../.. python agent.py # --- everything below uses the `evalshift` binary from the CLI --- diff --git a/llms-full.txt b/llms-full.txt index 6d32098..614badd 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -12,9 +12,18 @@ Object stores: pip install boto3 (s3://) | google-cloud-storage (gs://) | azure- Purpose: in-process capture SDK for AI agents. Wraps agent invocations, tool calls, and model calls; records each run as a span tree; serializes to a JSON capture envelope written to /captures//cap_.json on local disk, or (EVALSHIFT_CAPTURE_STORE=s3://|gs://|az://...) -to an object store the user owns -- never to EvalShift. +to an object store the user owns -- never to EvalShift. A capture.agent_session / +agent_session_async turn with a conversation_id goes to +captures//conv_<16 hex>/cap_.json instead (hex = start of sha256(conversation_id)); +every other capture, including every @capture.agent decorator capture, stays flat. +EVALSHIFT_CAPTURE_LAYOUT=flat (or layout="flat" on FileSink / ObjectStoreSink) writes every +capture flat, as before 0.7.0. If an evalshift CLI older than 1.6 is installed in the same +environment, the SDK writes flat and logs a warning (EVALSHIFT_CAPTURE_LAYOUT=conversation +overrides it); an older CLI in a different environment (e.g. CI pinned to 1.5) can't be +detected -- upgrade it first. Captures are consumed by the separate evalshift CLI -(https://github.com/evalshift/evalshift-cli); the layout (captures//cap_.json + +(https://github.com/evalshift/evalshift-cli), which reads the folders from 1.6.0; the layout +(captures//cap_.json + captures//conv_<16 hex>/cap_.json + toolsets/.json) is the interface between SDK and CLI, on disk or in a bucket. Quickstart (minimum working setup): @@ -28,8 +37,10 @@ Quickstart (minimum working setup): @capture.tool. 3. Run with EVALSHIFT_CAPTURE=1 set (capture is OFF by default — without it every wrapper is a pure pass-through and no file is written). -4. Result: .evalshift/captures//cap_.json appears in the CWD, ready for the - evalshift CLI. Full runnable code in "Minimal examples" below. +4. Result: .evalshift/captures//cap_.json appears in the CWD (a decorator capture + is always flat; an agent_session turn with a conversation_id lands in + captures//conv_<16 hex>/), ready for the evalshift CLI. Full runnable code in + "Minimal examples" below. Safety model (four distinct rules): 1. CAPTURE PATH IS FAIL-OPEN: every piece of SDK bookkeeping is guarded; a fault drops the @@ -52,10 +63,11 @@ Safety model (four distinct rules): |---|---|---|---| | EVALSHIFT_CAPTURE | unset (OFF) | Master gate. Truthy set (case-insensitive, stripped): {"1","true","yes","on"}. Anything else = off. | Live, every call | | EVALSHIFT_DIR | ".evalshift" | Capture root dir. Relative paths resolve against CWD (no repo-root walk). | Live, each write | -| EVALSHIFT_MAX_CAPTURES | 200 | Keep newest N *.json per suite dir (GC after each disk write, ordered by file mtime). | Config construction (import / reset_config) | +| EVALSHIFT_MAX_CAPTURES | 200 | Keep newest N *.json per suite dir, by unit: a capture or a whole conversation folder (GC after each disk write, ordered by file mtime). | Config construction (import / reset_config) | | EVALSHIFT_CAPTURE_TTL | off | Evict capture files older than N seconds. | Config construction | | EVALSHIFT_DEDUP | on | Per-process dedup keyed (suite, input_hash). | Config construction | -| EVALSHIFT_SAMPLE_RATE | off (capture all) | Fraction of runs to capture (0.0-1.0), decided once per agent invocation. | Config construction | +| EVALSHIFT_SAMPLE_RATE | off (capture all) | Fraction of runs to capture (0.0-1.0), decided once per agent invocation; per conversation when an agent_session is given a conversation_id. | Config construction | +| EVALSHIFT_CAPTURE_LAYOUT | unset (conversation folders) | "flat" (any case, stripped) writes every capture flat, as before 0.7.0; "conversation" writes conversation folders even when an evalshift CLI older than 1.6 is installed in the same environment. Unset or any other value means conversation folders, unless such an older CLI is installed alongside: then flat, with ONE warning per process. A layout= argument on FileSink / ObjectStoreSink wins over it. | Live, each write (installed-CLI check once per process) | | EVALSHIFT_CAPTURE_STORE | unset (disk) | Ship captures to s3:///, gs:/// or az:////; needs the provider's client library (pip install boto3 / google-cloud-storage / azure-storage-blob azure-identity). Unusable value (library missing, bad URI) -> recorded at import, raised as SinkConfigurationError by the first capture.* decorator / client wrapper / LangChain handler / configure() while EVALSHIFT_CAPTURE is on; never on bare import, never with capture off; if the gate turns on later, captures are dropped with ONE warning (never written to disk instead). Message names the var, scheme and package, never the value. EVALSHIFT_SINK (0.5.0 name) still works when this is unset. | Config construction | "Uncapped/off" literal set for numeric knobs: {"0","none","unlimited","off"} (also empty string). @@ -90,7 +102,9 @@ capture.agent(*, suite: str, redact: Redactor | bool, tools: Any, code_version: re-read at every call, not frozen at decoration — enabling EVALSHIFT_CAPTURE after import works). Agent input auto-derived by binding call args to the signature ({param: value}); binding failure falls back to {"args": [...], "kwargs": {...}}. Conversation kwargs are STATIC - (fixed at decoration time) — use agent_session for per-turn values. + (fixed at decoration time) — use agent_session for per-turn values. A static conversation_id + names no conversation, so decorator captures are ALWAYS written flat and sampled at random per + run, even with one; the envelope still carries the id. capture.agent_session(*, suite: str, redact: Redactor | bool, tools: Any, code_version: str = "", agent_input: Any = None, conversation_id: str | None = None, @@ -104,6 +118,8 @@ capture.agent_session(*, suite: str, redact: Redactor | bool, tools: Any, code_v rules): any JSON-able value; it is identity only — hashed as-is into the envelope input_hash (dedup key), never stored raw, and redaction does not apply to it. Recommended value: the full messages list. Recommended primitive for multi-turn conversations (fresh turn_index per with). + With a conversation_id: written to captures//conv_<16 hex>/, and a sample rate keeps or + drops the conversation whole (hash of the id, same answer in every process). capture.agent_session_async(...) Identical params/behavior; `async with` form. @@ -227,9 +243,16 @@ default_redactor(value: Any) -> Any Sink (protocol, @runtime_checkable, import from evalshift.sinks): write(envelope: CaptureEnvelope) -> Path | None (CaptureEnvelope imports from evalshift.trace) -FileSink(base: str | os.PathLike[str] | None = None) +FileSink(base: str | os.PathLike[str] | None = None, *, + layout: Literal["conversation", "flat"] | None = None) .write(envelope) -> Path | None. Writes /captures//.json (UTF-8 - JSON). Base resolution AT WRITE TIME: constructor arg > EVALSHIFT_DIR > ".evalshift" relative + JSON), or /captures//conv_<16 hex>/.json for an agent_session turn + with a conversation_id (decorator captures always flat). layout="flat" or + EVALSHIFT_CAPTURE_LAYOUT=flat -> every capture flat; layout="conversation" or + EVALSHIFT_CAPTURE_LAYOUT=conversation -> folders; the argument wins over the env var. With + neither: folders, unless an evalshift CLI older than 1.6 is installed in the same environment + (then flat, ONE warning per process). + Base resolution AT WRITE TIME: constructor arg > EVALSHIFT_DIR > ".evalshift" relative to CWD. Returns absolute Path, or None on OSError (capture dropped, debug log). Suite segment sanitized against path traversal (separators and ".." replaced). @@ -243,8 +266,11 @@ MemorySink() every capture's toolset_ref stays unstamped and the CLI refuses to promote it. With an ObjectStoreSink the sidecar goes to the store instead. -ObjectStoreSink(store: ObjectStore, *, background: bool = True, queue_size: int = 1000, flush_timeout: float = 10.0) - .write(envelope) -> None (keys: captures//.json; enqueued, daemon +ObjectStoreSink(store: ObjectStore, *, background: bool = True, queue_size: int = 1000, flush_timeout: float = 10.0, + layout: Literal["conversation", "flat"] | None = None) + .write(envelope) -> None (keys: captures//.json, or + captures//conv_<16 hex>/.json for an agent_session turn with a + conversation_id -- the FileSink layout rule, same layout= / EVALSHIFT_CAPTURE_LAYOUT; enqueued, daemon thread uploads; queue full -> newest dropped, debug line). .write_toolset(normalized, fingerprint) -> str | None (toolsets/.json, once per process; sidecar goes to the SAME store, replacing the file-based ToolsetSink; None when not accepted -> toolset_ref left @@ -257,7 +283,8 @@ ObjectStoreSink(store: ObjectStore, *, background: bool = True, queue_size: int sys.exit(0)) or call evalshift.flush_captures(timeout) in your shutdown hook (never inside a signal handler: it can deadlock on the sink's lock); on Lambda call flush_captures() before returning or use background=False. No retry layer (clients retry). - GC/max_captures/capture_ttl do not apply -> bucket lifecycle rules. Dedup still applies. + GC/max_captures/capture_ttl do not apply -> bucket lifecycle rules (a rule expires each object + by its own age, so a live conversation's oldest turns can expire first). Dedup still applies. ObjectStore (protocol, evalshift.stores): put(key: str, data: bytes) -> None Stores (lazy client, built on first put): evalshift.stores.s3.S3Store(bucket, prefix="", *, client=None) [AWS_ENDPOINT_URL honoured -> MinIO/R2/B2]; evalshift.stores.gcs.GCSStore(bucket, @@ -398,11 +425,13 @@ evalshift.capture.requested.extract_requested_tool_calls(response: Any) -> list[ pass agent_input, or set conversation_id. - conversation_id set -> input_hash = hash({agent_input, conversation_id, turn_index}), so repeated short turns ("yes", "1pm") don't dedup-collapse. conversation_id=None -> input_hash = - hash(agent_input), byte-identical to pre-1.1.0. + hash(agent_input), byte-identical to pre-1.1.0. Conversation folders change nothing here: + dedup never drops a distinct turn. - The SDK never returns the written capture_id to the caller; parent_capture_id must be user-managed or omitted. - @capture.agent conversation kwargs are static per decoration; per-turn values require - agent_session / agent_session_async. + agent_session / agent_session_async. Decorator captures are always written flat and sampled at + random per run, even with a conversation_id. - Tool span -> 2 events (tool_call + tool_result). model_call/error spans -> 1 event each. - Event ordering: sort by (timestamp, monotonic op-order) -> dense sequence_index; deterministic under concurrency. Span timing/parentage stored under event.metadata["evalshift"]. @@ -419,7 +448,22 @@ evalshift.capture.requested.extract_requested_tool_calls(response: Any) -> list[ model_call's output. - Hygiene defaults: dedup ON, max_captures 200/suite, TTL off, sampling off. Escape hatch: EVALSHIFT_MAX_CAPTURES=0 EVALSHIFT_DEDUP=off. GC runs only after a real disk write (Path - returned), orders by file mtime, never recurses, never raises. + returned), per suite dir, by unit: a capture or a whole conversation folder (conv_<16 hex>/, + timed by its newest file's mtime). TTL rule first: every file older than capture_ttl goes (an + idle conversation goes whole, a live one loses only old turns). Count rule on the files the TTL + leaves: units newest first (by newest surviving file), files within a unit newest first, keep + the first max_captures. So the newest conversations are kept whole while they fit, the one at + the boundary keeps its newest turns, older ones are evicted whole; one newer capture evicts at + most one older capture, not a whole live conversation; a never-ending conversation (a user id as conversation_id) + keeps only its newest max_captures turns. Emptied folders removed. A folder or file GC cannot + read is skipped (never counted, never removed). Flat-only suites behave exactly as before 0.7.0. + Enters only conv_* folders, one level; never raises. +- Sampling: decided once per agent invocation; per conversation when an agent_session / + agent_session_async is given a conversation_id (draw = sha256("evalshift-sample:" + id), so + every turn is kept or dropped together in every process). The conversation_id is what gets + sampled: a user id passed as conversation_id samples users, and a constant id in agent_session + keeps every run or none. The decorator, runs without an id + and the LangChain handler draw random.random(). A fault in the draw -> capture. - require_model_call=True (opt-in): captures with no model_call event are dropped before redaction/serialization (debug log only). - Redaction is REQUIRED at every capture point (agent decorator / agent_session / @@ -435,8 +479,9 @@ evalshift.capture.requested.extract_requested_tool_calls(response: Any) -> list[ logging.getLogger("evalshift") debug line. Gate-off, sampling, and dedup drops are silent. - Deliberate WARNING-level exceptions to debug-only logging: an invalid/unusable EVALSHIFT_CAPTURE_STORE (once, at config construction), the first failed put per ObjectStoreSink, an ObjectStoreSink - exit flush that times out with items pending, and a newer-minor schema read by - load_capture/load_envelope. + exit flush that times out with items pending, a newer-minor schema read by + load_capture/load_envelope, and (once per process) writing flat because an evalshift CLI older + than 1.6 is installed in the same environment. - retrieval / guardrail / final_output event types exist in the schema but have NO public recording API in the manual surface; only the LangChain adapter (retrieval, final_output) and internals emit them. Do not document them as user-recordable. @@ -444,12 +489,22 @@ evalshift.capture.requested.extract_requested_tool_calls(response: Any) -> list[ distinct fingerprint by ToolsetSink (with an ObjectStoreSink: toolsets/.json in the store, once per process); a repeat write of the same toolset is a cheap no-op existence check, not a second write. GC (hygiene/gc.py) never touches this directory: its - caller always passes /captures//, a sibling, and evict() never recurses into - subdirectories regardless. No orphan sweeping exists for toolsets in this SDK. + caller always passes /captures//, a sibling, and evict() enters only + conv_<16 hex> conversation folders, one level, never any other subdirectory. No orphan + sweeping exists for toolsets in this SDK. An evalshift CLI older than 1.6.0 does not see + folder captures, so its `capture clean` can sweep sidecars only they use -- upgrade the CLI + first. ## Capture file format Path: /captures//.json (capture_id = "cap_" + uuid4 hex) + or /captures//conv_<16 hex>/.json for an agent_session / + agent_session_async turn with a conversation_id; conv_ + the first 16 lowercase hex of + sha256(str(conversation_id)), never the raw id (readers match conv_[0-9a-f]{16}). Decorator + captures are always flat. EVALSHIFT_CAPTURE_LAYOUT=flat / layout="flat" -> every capture flat, + as before 0.7.0; so does an evalshift CLI older than 1.6 installed in the same environment, + unless EVALSHIFT_CAPTURE_LAYOUT=conversation / layout="conversation". Same relative layout under a store's prefix. The folder is not an envelope + field: the schema is unchanged. Toolset sidecars: /toolsets/.json (see ToolsetSink above) Envelope keys (unchanged since schema 1.1.0, in order): schema_version, capture_id, suite, @@ -544,6 +599,7 @@ for i, user_text in enumerate(turns): reply = run_model(messages) record_model_call(model_id="claude-sonnet-5", tools=[], input=messages, output=reply) messages.append({"role": "assistant", "content": reply}) +# Every turn -> .evalshift/captures/scheduler/conv_<16 hex of sha256(conv)>/cap_.json # 4. LangChain (pip install "evalshift-sdk[langchain]") from evalshift.adapters.langchain import EvalShiftCallbackHandler @@ -609,7 +665,8 @@ except MigrationError as e: # read path raises; it does NOT fail ## Troubleshooting: no file written Check in order: (1) EVALSHIFT_CAPTURE not truthy; (2) wrong CWD — default .evalshift is -CWD-relative, set EVALSHIFT_DIR; (3) dedup collapsed it (classic: agent_session without +CWD-relative, set EVALSHIFT_DIR (an agent_session turn with a conversation_id is one level +down, in captures//conv_<16 hex>/); (3) dedup collapsed it (classic: agent_session without agent_input); (4) sampling skipped it; (5) require_model_call dropped it; (6) redactor raised (fail-closed); (7) filesystem OSError (use MemorySink on read-only mounts); (8) ObjectStoreSink dropped it -- nothing local, check the bucket and the logs: first failed put per sink -> WARNING diff --git a/llms.txt b/llms.txt index 742a64f..eecb39d 100644 --- a/llms.txt +++ b/llms.txt @@ -1,6 +1,6 @@ # evalshift-sdk -> In-process Python capture SDK for AI agents. Wrap an agent's entry point, tool calls, and model calls; each run is recorded as a structured trace and written as a JSON capture file to `.evalshift/captures//cap_.json` on local disk. Capture is off unless the `EVALSHIFT_CAPTURE` env var is truthy. Zero runtime dependencies (stdlib only), Python >= 3.10, no network code unless `EVALSHIFT_CAPTURE_STORE` names an object store you own. Captures are consumed by the separate [evalshift CLI](https://github.com/evalshift/evalshift-cli); the capture layout is the interface between SDK and CLI, on disk or in a bucket. +> In-process Python capture SDK for AI agents. Wrap an agent's entry point, tool calls, and model calls; each run is recorded as a structured trace and written as a JSON capture file to `.evalshift/captures//cap_.json` on local disk, or to `.evalshift/captures//conv_<16 hex>/cap_.json` for a `capture.agent_session` / `agent_session_async` turn with a `conversation_id` (the hex is the start of `sha256(conversation_id)`; `@capture.agent` decorator captures are always flat; `EVALSHIFT_CAPTURE_LAYOUT=flat` writes every capture flat, as before 0.7.0; reading the folders needs evalshift CLI 1.6.0 or later, and an older evalshift CLI installed in the same environment makes the SDK write flat with a warning unless `EVALSHIFT_CAPTURE_LAYOUT=conversation`). Capture is off unless the `EVALSHIFT_CAPTURE` env var is truthy. Zero runtime dependencies (stdlib only), Python >= 3.10, no network code unless `EVALSHIFT_CAPTURE_STORE` names an object store you own. Captures are consumed by the separate [evalshift CLI](https://github.com/evalshift/evalshift-cli); the capture layout is the interface between SDK and CLI, on disk or in a bucket. Key facts: @@ -8,7 +8,7 @@ Key facts: - Top-level public API: `capture` (with `.agent`, `.agent_session`, `.agent_session_async`, `.model_call`, `.tool`), `record_model_call`, `configure`, `Redactor`, `RedactSetting`, `default_redactor`, `FileSink`, `MemorySink`, `ObjectStoreSink`, `flush_captures`, `load_capture`, `load_envelope`, `register_migration`, `MigrationError`, `SCHEMA_VERSION`, `SinkConfigurationError`, `__version__`. LangChain adapter: `evalshift.adapters.langchain.EvalShiftCallbackHandler` (needs the `[langchain]` extra). Provider client wrappers: `evalshift.adapters.openai.wrap_openai`, `evalshift.adapters.anthropic.wrap_anthropic`, `evalshift.adapters.genai.wrap_genai` (extras `[openai]`, `[anthropic]`, `[google-genai]`) — a drop-in proxy over a client instance that records one `model_call` per request inside a capture session. Object stores: `evalshift.stores` (`ObjectStore`, `MemoryStore`, `open_store`, `parse_store_uri`) plus `evalshift.stores.s3.S3Store` / `.gcs.GCSStore` / `.azure.AzureBlobStore` — each needs the provider's client library (`pip install boto3` / `google-cloud-storage` / `azure-storage-blob azure-identity`), no extra needed. - Redaction is required, not opt-in: `redact=` is a mandatory keyword on `capture.agent`, `capture.agent_session`, `capture.agent_session_async`, and `EvalShiftCallbackHandler` — `True` masks with `default_redactor`, `False` captures verbatim, a callable is your own redactor; anything else (including `None`) raises `TypeError`. There is no `configure(redact=...)`. - Safety model: the capture path is fail-open (SDK faults can never break or slow the host agent; failed runs are still captured, then the original exception re-raises); redaction is fail-closed (a raising redactor drops the capture rather than writing unredacted data); the read path (`load_capture`/`load_envelope`) raises typed `MigrationError` subclasses. -- Hygiene defaults keep `captures/` bounded: dedup on (per-process, keyed `(suite, input_hash)`), newest-200 files kept per suite. Configurable via `configure(...)` or `EVALSHIFT_MAX_CAPTURES` / `EVALSHIFT_DEDUP` / `EVALSHIFT_CAPTURE_TTL` / `EVALSHIFT_SAMPLE_RATE`; capture root via `EVALSHIFT_DIR`. +- Hygiene defaults keep `captures/` bounded: dedup on (per-process, keyed `(suite, input_hash)`), newest-200 files kept per suite dir (of those the TTL leaves), by unit: a capture or a whole conversation folder; the newest conversations are kept whole while they fit, the one at the boundary keeps its newest turns, older ones are evicted whole. Sampling, when set, is decided once per agent invocation; per conversation when an `agent_session` is given a `conversation_id`. Configurable via `configure(...)` or `EVALSHIFT_MAX_CAPTURES` / `EVALSHIFT_DEDUP` / `EVALSHIFT_CAPTURE_TTL` / `EVALSHIFT_SAMPLE_RATE`; capture root via `EVALSHIFT_DIR`. ## Docs diff --git a/src/evalshift/adapters/langchain.py b/src/evalshift/adapters/langchain.py index 58a6a43..fda87d3 100644 --- a/src/evalshift/adapters/langchain.py +++ b/src/evalshift/adapters/langchain.py @@ -302,6 +302,7 @@ def _register_run_locked(self, run_id: UUID, parent_run_id: UUID | None) -> UUID return root def _begin_session_locked(self, root: UUID, agent_input: Any) -> None: + # no conversation id here: sampled per run enabled = safety.guard( "langchain gate check", lambda: config.is_capture_enabled() and config.should_capture_now(), diff --git a/src/evalshift/capture/api.py b/src/evalshift/capture/api.py index 7375ed4..3d23983 100644 --- a/src/evalshift/capture/api.py +++ b/src/evalshift/capture/api.py @@ -29,6 +29,7 @@ from evalshift.capture.span import SpanTree, is_persistable from evalshift.capture.toolset import fingerprint_tools, normalize_tools from evalshift.redaction import Redactor, RedactSetting, resolve_redactor +from evalshift.sinks._layout import writing_flat from evalshift.trace.serialize import build_capture P = ParamSpec("P") @@ -255,6 +256,7 @@ def _finalize( conversation_id: str | None = None, turn_index: int | None = None, parent_capture_id: str | None = None, + flat: bool = False, ) -> None: # Opt-in persistence gate (configure(require_model_call=True)): a capture with no # model_call span carries no scoreable ground truth and cannot be promoted by the CLI @@ -281,7 +283,9 @@ def _finalize( ) if envelope is None: return - with safety.fail_open("sink write"): + # ``flat``: write outside any conversation folder (the decorator path, R10). The fail-open + # guard stays outermost so nothing in the layout override can raise into the host. + with safety.fail_open("sink write"), writing_flat() if flat else nullcontext(): config.active_sink().write(envelope) @@ -470,6 +474,7 @@ def _run_agent( turn_index: int | None = None, parent_capture_id: str | None = None, ) -> R: + # the decorator's conversation_id is static: sampled per run (D4) if not config.should_capture_now(): return fn(*args, **kwargs) # not sampled -> transparent pass-through (no tree built) tree = safety.guard("open session", SpanTree) @@ -493,6 +498,8 @@ def _run_agent( conversation_id=conversation_id, turn_index=turn_index, parent_capture_id=parent_capture_id, + # the decorator's conversation_id is static: it names no conversation (D1/D4) + flat=True, ) raise else: @@ -506,6 +513,8 @@ def _run_agent( conversation_id=conversation_id, turn_index=turn_index, parent_capture_id=parent_capture_id, + # the decorator's conversation_id is static: it names no conversation (D1/D4) + flat=True, ) return result @@ -525,6 +534,7 @@ async def _run_agent_async( ) -> Any: # Async mirror of _run_agent. _record_error/_finalize are sync (Sink.write is sync), so the # async path reuses them verbatim — no await. Same fail-open + error-event-then-raise contract. + # the decorator's conversation_id is static: sampled per run (D4) if not config.should_capture_now(): return await fn(*args, **kwargs) # not sampled -> transparent pass-through (no tree) tree = safety.guard("open session", SpanTree) @@ -548,6 +558,8 @@ async def _run_agent_async( conversation_id=conversation_id, turn_index=turn_index, parent_capture_id=parent_capture_id, + # the decorator's conversation_id is static: it names no conversation (D1/D4) + flat=True, ) raise else: @@ -561,6 +573,8 @@ async def _run_agent_async( conversation_id=conversation_id, turn_index=turn_index, parent_capture_id=parent_capture_id, + # the decorator's conversation_id is static: it names no conversation (D1/D4) + flat=True, ) return result @@ -845,7 +859,7 @@ def agent_session( if not config.is_capture_enabled(): yield None return - if not config.should_capture_now(): + if not config.should_capture_now(conversation_id): yield None return tree = safety.guard("open session", SpanTree) @@ -912,7 +926,7 @@ async def agent_session_async( if not config.is_capture_enabled(): yield None return - if not config.should_capture_now(): + if not config.should_capture_now(conversation_id): yield None return tree = safety.guard("open session", SpanTree) diff --git a/src/evalshift/config.py b/src/evalshift/config.py index 24ef25f..86cf25a 100644 --- a/src/evalshift/config.py +++ b/src/evalshift/config.py @@ -393,13 +393,17 @@ def flush_captures(timeout: float | None = None) -> bool: return True -def should_capture_now() -> bool: +def should_capture_now(conversation_id: str | None = None) -> bool: """Sampling decision for one agent run (read at agent entry). Fail-open: capture on fault. - A configured ``sample_rate`` decides the draw; a fault in the draw defaults to **capturing** - so a sampling bug never silently disables all telemetry. + A configured ``sample_rate`` decides the draw -- per conversation when ``conversation_id`` is + known, so its turns are kept or dropped together; a fault in the draw defaults to + **capturing** so a sampling bug never silently disables all telemetry. """ - decision = safety.guard("sample decision", lambda: should_capture(_CONFIG.sample_rate)) + decision = safety.guard( + "sample decision", + lambda: should_capture(_CONFIG.sample_rate, key=conversation_id), + ) return decision is not False diff --git a/src/evalshift/hygiene/__init__.py b/src/evalshift/hygiene/__init__.py index c00f7fc..b9d7c72 100644 --- a/src/evalshift/hygiene/__init__.py +++ b/src/evalshift/hygiene/__init__.py @@ -1,11 +1,15 @@ """Capture hygiene: sampling, dedup, and garbage collection (Phase 6). -These controls keep a busy agent's capture firehose self-managing. Each is **opt-in** and -**off by default** — with no hygiene knob set, capture behaves exactly as in Phases 0-5. +These controls keep a busy agent's capture firehose self-managing. Dedup and a ``max_captures`` +cap of 200 per suite directory are **on by default**; ``sample_rate`` and ``capture_ttl`` are +opt-in (``docs/DECISIONS.md`` D-6a). ``EVALSHIFT_MAX_CAPTURES=0 EVALSHIFT_DEDUP=off`` restores +unbounded capture. -* :mod:`evalshift.hygiene.sample` — capture a fraction of agent runs (decided at agent entry). +* :mod:`evalshift.hygiene.sample` — capture a fraction of agent runs, decided once per agent + invocation; per conversation when an ``agent_session`` is given a ``conversation_id``. * :mod:`evalshift.hygiene.dedup` — collapse captures with identical inputs (per-process registry). -* :mod:`evalshift.hygiene.gc` — cap a suite directory by count / age, evicting the oldest. +* :mod:`evalshift.hygiene.gc` — cap a suite directory by count / age, by unit: a capture or a + whole conversation folder, evicting the oldest. Stdlib only (D-deps). """ diff --git a/src/evalshift/hygiene/gc.py b/src/evalshift/hygiene/gc.py index caba13d..9f4bb61 100644 --- a/src/evalshift/hygiene/gc.py +++ b/src/evalshift/hygiene/gc.py @@ -1,16 +1,29 @@ """Capture garbage collection: cap a suite directory by count and/or age. -:func:`evict` runs after a successful disk write, on the suite directory just written to. It does -a single :func:`os.scandir` pass and orders by **filesystem mtime** (not the envelope's -``created_at``) — stat only, no JSON parsing, so it stays cheap even when ``max_captures`` is -large. Two rules combine (union): - -* **count** — keep the newest ``max_captures`` ``*.json`` files, evict the rest; -* **TTL** — evict any ``*.json`` whose age (``now_ts - mtime``) exceeds ``capture_ttl_seconds``. - -Each unlink is individually swallowed, and the whole scan is guarded, so eviction never raises -into the host. Limitation (``docs/DECISIONS.md`` D-6): tools that rewrite mtimes (``cp -p``, -rsync, tar-extract) can perturb eviction order. +:func:`evict` runs after a successful disk write, on the **suite** directory just written to. It +caps it by **unit**: a flat ``*.json`` capture, or a whole conversation folder +(``conv_<16 hex>/``, see :mod:`evalshift.sinks._layout`) timed by its newest file. It does a single +:func:`os.scandir` pass over the suite directory (plus one per conversation folder) and orders by +**filesystem mtime** (not the envelope's ``created_at``) — stat only, no JSON parsing, so it stays +cheap even when ``max_captures`` is large. Two rules apply, TTL first: + +* **TTL** — evict any ``*.json`` file whose age (``now_ts - mtime``) exceeds + ``capture_ttl_seconds``, so an idle conversation goes whole and a live one loses only its old + turns; +* **count** — on the files the TTL leaves: order units newest first by their newest surviving + file and the files within a unit newest first, then keep the first ``max_captures`` files in + that order. The newest units are kept whole while they fit, the unit at the boundary keeps its + newest turns that fit, and every older unit goes whole (so the newest unit, if it is larger + than the cap, keeps its newest ``max_captures`` turns). A flat-only suite keeps its newest + ``max_captures``. + +A conversation folder left empty is removed. Other subdirectories are never entered. An entry +that cannot be read (a flat file's stat, a folder's listing) is skipped -- never counted, never +removed -- so one unreadable or vanished folder cannot stop the rest of the GC. Each unlink is +individually swallowed, and the suite-directory scan is guarded, so eviction never raises into +the host. +Limitation (``docs/DECISIONS.md`` D-6): tools that rewrite mtimes (``cp -p``, rsync, +tar-extract) can perturb eviction order. Stdlib only (D-deps). """ @@ -18,9 +31,55 @@ from __future__ import annotations import os +from dataclasses import dataclass from pathlib import Path from evalshift.safety import logger +from evalshift.sinks._layout import is_conversation_dir + + +@dataclass(frozen=True, slots=True) +class _Unit: + """What GC counts together: one flat capture, or a conversation folder. + + ``files`` are ``(mtime, path)`` pairs, oldest first; the unit's time is its newest file's. + """ + + files: tuple[tuple[float, Path], ...] + folder: Path | None + + @property + def newest(self) -> float: + return self.files[-1][0] + + +def _scan_unit(entry: os.DirEntry[str]) -> _Unit | None: + """The unit ``entry`` is, or ``None`` for anything GC leaves alone (or a folder now empty).""" + if entry.is_file() and entry.name.endswith(".json"): + return _Unit(((entry.stat().st_mtime, Path(entry.path)),), None) + if entry.is_dir() and is_conversation_dir(entry.name): + with os.scandir(entry.path) as turns: + files = sorted( + (turn.stat().st_mtime, Path(turn.path)) + for turn in turns + if turn.is_file() and turn.name.endswith(".json") + ) + return _Unit(tuple(files), Path(entry.path)) if files else None + return None + + +def _scan(directory: Path) -> list[_Unit]: + units: list[_Unit] = [] + with os.scandir(directory) as entries: + for entry in entries: + try: + unit = _scan_unit(entry) + except OSError: # unreadable or vanished: skipping it only lowers the count + logger.debug("evalshift: gc skipped %s", entry.path, exc_info=True) + continue + if unit is not None: + units.append(unit) + return sorted(units, key=lambda u: (u.newest, u.files[0][1])) # oldest first def evict( @@ -30,42 +89,63 @@ def evict( capture_ttl_seconds: float | None, now_ts: float, ) -> list[Path]: - """Evict aged / overflow ``*.json`` captures from ``directory``; return the paths removed. - - No-op (returns ``[]``) when neither threshold is set or the directory is unreadable. Never - recurses into subdirectories; never raises into the caller. + """Evict aged / overflow captures from the suite ``directory``; return the files removed. + + TTL rule first: every file older than ``capture_ttl_seconds`` goes, so an idle conversation + goes whole and a live one loses only its old turns. Count rule on what the TTL leaves: units + ordered newest first (by their newest surviving file), files within a unit newest first, and + the first ``max_captures`` files in that order are kept -- the newest units whole while they + fit, the unit at the boundary its newest turns that fit, every older unit none. A folder left + empty is removed. An entry that cannot be read is skipped (never counted, never removed). No-op + (returns ``[]``) when neither threshold is set or the directory is unreadable; never raises + into the caller. """ if max_captures is None and capture_ttl_seconds is None: return [] try: - entries = [ - (entry.stat().st_mtime, Path(entry.path)) - for entry in os.scandir(directory) - if entry.is_file() and entry.name.endswith(".json") - ] + units = _scan(directory) except OSError: logger.debug("evalshift: gc scan of %s failed (skipped)", directory, exc_info=True) return [] - entries.sort(key=lambda item: item[0]) # oldest first doomed: set[Path] = set() - if capture_ttl_seconds is not None: - doomed.update(path for mtime, path in entries if now_ts - mtime > capture_ttl_seconds) - - if max_captures is not None and len(entries) > max_captures: - keep = {path for _, path in entries[len(entries) - max_captures :]} - doomed.update(path for _, path in entries if path not in keep) + doomed.update( + path + for unit in units + for mtime, path in unit.files + if now_ts - mtime > capture_ttl_seconds + ) + if max_captures is not None: + survivors = [ + files for unit in units if (files := tuple(f for f in unit.files if f[1] not in doomed)) + ] + survivors.sort(key=lambda files: (files[-1][0], files[0][1]), reverse=True) # newest first + kept = 0 + for files in survivors: + for _, path in reversed(files): # newest turn first + if kept < max_captures: + kept += 1 + else: + doomed.add(path) removed: list[Path] = [] - for _, path in entries: # oldest-first, for a stable removal order - if path not in doomed: - continue - try: - path.unlink(missing_ok=True) - removed.append(path) - except OSError: - logger.debug("evalshift: gc unlink of %s failed (skipped)", path, exc_info=True) + for unit in units: # oldest-first, for a stable removal order + hit = False + for _, path in unit.files: + if path not in doomed: + continue + hit = True + try: + path.unlink(missing_ok=True) + removed.append(path) + except OSError: + logger.debug("evalshift: gc unlink of %s failed (skipped)", path, exc_info=True) + if hit and unit.folder is not None: + try: + unit.folder.rmdir() # fails, harmlessly, while turns remain + except OSError: + logger.debug("evalshift: gc kept %s (not empty)", unit.folder) return removed diff --git a/src/evalshift/hygiene/sample.py b/src/evalshift/hygiene/sample.py index 1d26282..a1f733c 100644 --- a/src/evalshift/hygiene/sample.py +++ b/src/evalshift/hygiene/sample.py @@ -2,22 +2,38 @@ The decision is made **once, at agent entry** (before any span tree is built), so an unsampled run degrades to the same near-pure pass-through as the gate-off path — no tree, no redaction, no -serialization. The draw source ``rng`` is injectable so tests are deterministic. +serialization. With a conversation id the draw is a hash of it, so a conversation is captured +whole or not at all, in every process. The draw source ``rng`` is injectable so tests are +deterministic. Stdlib only (D-deps). """ from __future__ import annotations +import hashlib import random from collections.abc import Callable -def should_capture(rate: float | None, *, rng: Callable[[], float] | None = None) -> bool: +def conversation_draw(key: str) -> float: + """A stable draw in ``[0, 1)`` for ``key``: 64 bits of ``sha256("evalshift-sample:" + key)``. + + Every turn of a conversation, in every process, gets the same draw, so a sample rate keeps or + drops a conversation whole. + """ + digest = hashlib.sha256(f"evalshift-sample:{key}".encode()).hexdigest() + return int(digest[:16], 16) / 2**64 + + +def should_capture( + rate: float | None, *, rng: Callable[[], float] | None = None, key: str | None = None +) -> bool: """True iff this run should be captured under ``rate`` (a probability in ``[0, 1]``). - ``None`` or ``rate >= 1`` always captures; ``rate <= 0`` never captures; otherwise capture - iff a uniform draw ``rng()`` falls below ``rate`` (strict ``<``, so ``rate == draw`` drops). + ``None`` or ``rate >= 1`` always captures; ``rate <= 0`` never captures. Otherwise, with a + ``key`` (a conversation id) capture iff :func:`conversation_draw` of it falls below ``rate``; + without one, iff a uniform draw ``rng()`` does (strict ``<``, so ``rate == draw`` drops). ``rng`` defaults to :func:`random.random`, resolved at call time so tests can monkeypatch it. """ @@ -25,8 +41,10 @@ def should_capture(rate: float | None, *, rng: Callable[[], float] | None = None return True if rate <= 0.0: return False + if key is not None: + return conversation_draw(key) < rate draw = rng if rng is not None else random.random return draw() < rate -__all__ = ["should_capture"] +__all__ = ["conversation_draw", "should_capture"] diff --git a/src/evalshift/sinks/_layout.py b/src/evalshift/sinks/_layout.py new file mode 100644 index 0000000..641b4c6 --- /dev/null +++ b/src/evalshift/sinks/_layout.py @@ -0,0 +1,147 @@ +"""Where a capture lives under ``captures/``: in its conversation's folder, or flat. + +A capture with a ``conversation_id`` goes to ``/conv_<16 hex>/.json``, +where the hex is the start of ``sha256(conversation_id)``: fixed length, lowercase, safe on every +filesystem, and never the raw id. Every other capture goes to ``/.json``. +The layout is the same under ``/captures/`` and under a store's prefix, so a bucket stays a +mirror of ``.evalshift/``. ``EVALSHIFT_CAPTURE_LAYOUT=flat`` (or ``layout="flat"`` on a sink) +writes every capture flat, as before 0.7.0. The evalshift CLI reads both layouts from 1.6.0. +Captures from the ``@capture.agent`` decorator always stay flat (:func:`writing_flat`): its +``conversation_id`` is fixed when the function is decorated, so it names no conversation. + +**Co-install guard.** An older evalshift CLI cannot see conversation folders, and a CLI 1.5.x +install can pull this SDK in without the user choosing it. So when neither a sink's ``layout=`` +nor the env var picks a layout, and an evalshift CLI older than 1.6 is installed in this +environment, captures are written flat and one warning is logged per process. +``EVALSHIFT_CAPTURE_LAYOUT=conversation`` opts in anyway. A CLI in another environment (CI pinned +to 1.5, say) cannot be detected. + +Stdlib only (D-deps). +""" + +from __future__ import annotations + +import functools +import hashlib +import logging +import os +import re +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from importlib import metadata +from typing import Literal + +CaptureLayout = Literal["conversation", "flat"] + +#: Env var selecting the layout: ``flat`` or ``conversation`` (any case, stripped). +LAYOUT_ENV = "EVALSHIFT_CAPTURE_LAYOUT" + +#: The first evalshift CLI (``major``, ``minor``) that reads conversation folders. +_FIRST_FOLDER_CLI = (1, 6) + +#: The leading ``major.minor`` of a version string; ``1.6.0rc1`` and ``1.6.0.dev0`` count as 1.6. +_VERSION = re.compile(r"(\d+)\.(\d+)") + +#: The SDK's logger (the one :mod:`evalshift.safety` uses), by name: this module is stdlib-only. +_log = logging.getLogger("evalshift") + +_CONVERSATION_DIR = re.compile(r"conv_[0-9a-f]{16}") + +_FLAT_WRITE: ContextVar[bool] = ContextVar("evalshift_flat_write", default=False) + + +def conversation_dir(conversation_id: str) -> str: + """The folder a conversation's turns live in: ``conv_`` + 16 hex digits of its sha256. + + ``str()`` first: the envelope types the id as ``str``, but nothing enforces it, and a stray + int must not turn into a dropped capture. + """ + return "conv_" + hashlib.sha256(str(conversation_id).encode("utf-8")).hexdigest()[:16] + + +def is_conversation_dir(name: str) -> bool: + """Whether ``name`` is a folder :func:`conversation_dir` produces.""" + return _CONVERSATION_DIR.fullmatch(name) is not None + + +@contextmanager +def writing_flat() -> Iterator[None]: + """Write the captures inside this block flat, whatever the layout says. + + The ``@capture.agent`` decorator wraps its sink write in this: its ``conversation_id`` is + fixed at decoration time, so it names no conversation and its runs never share a folder. + """ + token = _FLAT_WRITE.set(True) + try: + yield + finally: + _FLAT_WRITE.reset(token) + + +def _cli_version() -> str: + """The installed evalshift CLI's version; raises ``PackageNotFoundError`` when it is absent. + + The CLI distribution is ``evalshift`` (this SDK's is ``evalshift-sdk``). + """ + return metadata.version("evalshift") + + +@functools.cache +def _guard_layout() -> CaptureLayout: + """``flat`` when an evalshift CLI older than 1.6 is installed alongside, else ``conversation``. + + Computed once per process (``cache_clear()`` resets it), so the one warning is logged once. + Anything unreadable -- no CLI, an odd version string, a failing lookup -- means no guard. + """ + try: + version = _cli_version() + match = _VERSION.match(version) + if match is None or (int(match[1]), int(match[2])) >= _FIRST_FOLDER_CLI: + return "conversation" + _log.warning( + "evalshift: the installed evalshift CLI %s can't read conversation folders, so " + "captures are written flat; upgrade the CLI to 1.6.0 or set " + "EVALSHIFT_CAPTURE_LAYOUT=conversation", + version, + ) + return "flat" + except Exception: # never let the guard stop a capture + return "conversation" + + +def resolve_layout(explicit: CaptureLayout | None = None) -> CaptureLayout: + """The layout to write, first match wins. + + 1. ``flat`` inside :func:`writing_flat`; + 2. ``explicit``, when a sink was given ``layout=``; + 3. the env var, when it says ``flat`` or ``conversation`` (any case, stripped) -- so + ``conversation`` is the explicit opt-in that overrides the guard; + 4. else (unset, empty or a typo) the co-install guard: ``flat`` when an evalshift CLI older + than 1.6 is installed in this environment (one warning per process), else + ``conversation``. + + Never raises, so a bad value can never stop captures. A sink must call this synchronously + inside ``write()``: :func:`writing_flat` is a context variable, so a call made later (on a + background thread, say) would not see it. + """ + if _FLAT_WRITE.get(): + return "flat" + if explicit is not None: + return explicit + raw = os.environ.get(LAYOUT_ENV, "").strip().lower() + if raw == "flat": + return "flat" + if raw == "conversation": + return "conversation" + return _guard_layout() + + +__all__ = [ + "LAYOUT_ENV", + "CaptureLayout", + "conversation_dir", + "is_conversation_dir", + "resolve_layout", + "writing_flat", +] diff --git a/src/evalshift/sinks/file.py b/src/evalshift/sinks/file.py index 35104b8..5dbb333 100644 --- a/src/evalshift/sinks/file.py +++ b/src/evalshift/sinks/file.py @@ -1,6 +1,8 @@ """Basic on-disk sink: one ``cap_.json`` per capture. -Path layout is ``/captures//.json``. ``base`` resolves at write time: +Path layout is ``/captures//.json``, or +``/captures//conv_<16 hex>/.json`` for a conversation's turns (see +:mod:`evalshift.sinks._layout`). ``base`` resolves at write time: an explicit constructor argument wins, else the ``EVALSHIFT_DIR`` env var, else ``.evalshift`` relative to the current working directory. There is deliberately **no** repo-root walk — the SDK must not assume it runs inside a checkout (containers, Lambda). Resolution order lives in @@ -21,6 +23,7 @@ from pathlib import Path from evalshift.safety import logger +from evalshift.sinks._layout import CaptureLayout, conversation_dir, resolve_layout from evalshift.sinks._paths import resolve_base from evalshift.trace.models import CaptureEnvelope from evalshift.trace.serialize import capture_filename, dumps @@ -35,11 +38,26 @@ def _safe_segment(name: str) -> str: return cleaned or "_" +def capture_relpath(envelope: CaptureEnvelope, *, layout: CaptureLayout) -> str: + """``/[conv_<16 hex>/].json`` -- the same under ``captures/`` on + disk and in a store (see :mod:`evalshift.sinks._layout`).""" + suite = _safe_segment(envelope.suite) + name = capture_filename(envelope.capture_id) + if layout == "conversation" and envelope.conversation_id is not None: + return f"{suite}/{conversation_dir(envelope.conversation_id)}/{name}" + return f"{suite}/{name}" + + class FileSink: - """Write capture envelopes to ``/captures//.json``.""" + """Write capture envelopes to ``/captures//.json``, or + ``/captures//conv_<16 hex>/.json`` for a conversation's turns + (see :mod:`evalshift.sinks._layout`).""" - def __init__(self, base: str | os.PathLike[str] | None = None) -> None: + def __init__( + self, base: str | os.PathLike[str] | None = None, *, layout: CaptureLayout | None = None + ) -> None: self._base = base + self._layout = layout @property def base(self) -> str | os.PathLike[str] | None: @@ -56,6 +74,10 @@ def base(self) -> str | os.PathLike[str] | None: def _resolve_base(self) -> Path: return resolve_base(self._base) + def suite_dir(self, envelope: CaptureEnvelope) -> Path: + """``/captures/`` -- what GC caps, whichever folder a write landed in.""" + return (self._resolve_base() / "captures" / _safe_segment(envelope.suite)).absolute() + def write(self, envelope: CaptureEnvelope) -> Path | None: """Serialize ``envelope`` and write it; create parent dirs; return the path written. @@ -63,8 +85,12 @@ def write(self, envelope: CaptureEnvelope) -> Path | None: so callers get an unambiguous location regardless of how ``base`` was specified. Returns ``None`` when the write degrades on a filesystem ``OSError`` (read-only / disk full). """ - target_dir = (self._resolve_base() / "captures" / _safe_segment(envelope.suite)).absolute() - path = target_dir / capture_filename(envelope.capture_id) + path = ( + self._resolve_base() + / "captures" + / capture_relpath(envelope, layout=resolve_layout(self._layout)) + ).absolute() + target_dir = path.parent try: target_dir.mkdir(parents=True, exist_ok=True) path.write_text(dumps(envelope), encoding="utf-8") diff --git a/src/evalshift/sinks/hygiene.py b/src/evalshift/sinks/hygiene.py index 924617a..037cf17 100644 --- a/src/evalshift/sinks/hygiene.py +++ b/src/evalshift/sinks/hygiene.py @@ -25,6 +25,7 @@ from evalshift import safety from evalshift.hygiene import dedup as dedup_registry from evalshift.hygiene import gc as capture_gc +from evalshift.sinks._layout import is_conversation_dir from evalshift.sinks.base import Sink from evalshift.trace.models import CaptureEnvelope @@ -76,8 +77,15 @@ def write(self, envelope: CaptureEnvelope) -> Path | None: if path is not None and (self._max_captures is not None or self._capture_ttl is not None): with safety.fail_open("gc"): + suite_dir = getattr(self._base, "suite_dir", None) + if callable(suite_dir): + directory = suite_dir(envelope) + elif is_conversation_dir(path.parent.name): + directory = path.parent.parent # a custom sink that wrote a folder + else: + directory = path.parent capture_gc.evict( - path.parent, + directory, max_captures=self._max_captures, capture_ttl_seconds=self._capture_ttl, now_ts=self._now(), diff --git a/src/evalshift/sinks/object_store.py b/src/evalshift/sinks/object_store.py index 72de784..355ffd9 100644 --- a/src/evalshift/sinks/object_store.py +++ b/src/evalshift/sinks/object_store.py @@ -6,10 +6,12 @@ ``EVALSHIFT_CAPTURE_STORE=`` (see :mod:`evalshift.config`) or ``configure(sink=ObjectStoreSink(store))``. -**Key layout is the local layout.** ``captures//.json`` and -``toolsets/.json`` under the store's prefix -- byte for byte what -:class:`~evalshift.sinks.file.FileSink` and :class:`~evalshift.sinks.toolset.ToolsetSink` -write, so the CLI mirrors a bucket into ``.evalshift/`` and reads it unchanged. +**Key layout is the local layout.** ``captures//.json`` (or +``captures//conv_<16 hex>/.json`` for a conversation's turns, see +:mod:`evalshift.sinks._layout`) and ``toolsets/.json`` under the store's prefix -- byte for +byte what :class:`~evalshift.sinks.file.FileSink` and +:class:`~evalshift.sinks.toolset.ToolsetSink` write, so the CLI mirrors a bucket into +``.evalshift/`` and reads it unchanged. **Background by default.** ``write`` enqueues and returns; one daemon thread drains the queue with ``store.put``. The queue is bounded (``queue_size``); when it is full the incoming item is @@ -49,11 +51,12 @@ from typing import Any from evalshift.safety import logger -from evalshift.sinks.file import _safe_segment +from evalshift.sinks._layout import CaptureLayout, resolve_layout +from evalshift.sinks.file import capture_relpath from evalshift.sinks.toolset import toolset_payload from evalshift.stores.base import ObjectStore from evalshift.trace.models import CaptureEnvelope -from evalshift.trace.serialize import capture_filename, dumps +from evalshift.trace.serialize import dumps # (key, data, fingerprint): the fingerprint is set for toolset sidecars and ``None`` for captures, # so a failed background sidecar put can release it for retry. @@ -70,8 +73,10 @@ def __init__( background: bool = True, queue_size: int = 1000, flush_timeout: float = 10.0, + layout: CaptureLayout | None = None, ) -> None: self._store = store + self._layout = layout self._background = background self._flush_timeout = flush_timeout self._queue: queue.Queue[_Item] = queue.Queue(maxsize=queue_size) @@ -89,8 +94,10 @@ def store(self) -> ObjectStore: return self._store def write(self, envelope: CaptureEnvelope) -> Path | None: - """Ship ``envelope`` under ``captures//.json``; return ``None``.""" - key = f"captures/{_safe_segment(envelope.suite)}/{capture_filename(envelope.capture_id)}" + """Ship ``envelope`` under ``captures//.json``, or under + ``captures//conv_<16 hex>/.json`` for a conversation's turn; + return ``None``.""" + key = f"captures/{capture_relpath(envelope, layout=resolve_layout(self._layout))}" self._submit(key, dumps(envelope).encode("utf-8")) return None diff --git a/tests/conftest.py b/tests/conftest.py index 41be5f9..66863f9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,6 +2,7 @@ from __future__ import annotations +import importlib.metadata import json from collections.abc import Callable, Iterator from pathlib import Path @@ -10,6 +11,7 @@ import pytest from evalshift.config import reset_config +from evalshift.sinks import _layout CaptureReader = Callable[[str], list[dict[str, Any]]] @@ -21,7 +23,16 @@ def _isolate_config(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: The hygiene knobs now default from env vars, so clear them first: the test-suite baseline must be the built-in defaults (dedup on, max_captures=200) regardless of the developer's shell env. Tests that exercise a specific env value ``monkeypatch.setenv(...)`` then ``reset_config()``. + + The co-install guard (R11) must not depend on whether an evalshift CLI happens to be + installed: by default it sees none, so the layout stays ``conversation``. """ + + def _no_cli() -> str: + raise importlib.metadata.PackageNotFoundError("evalshift") + + monkeypatch.setattr(_layout, "_cli_version", _no_cli) + _layout._guard_layout.cache_clear() for var in ( "EVALSHIFT_MAX_CAPTURES", "EVALSHIFT_CAPTURE_TTL", @@ -29,11 +40,13 @@ def _isolate_config(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: "EVALSHIFT_SAMPLE_RATE", "EVALSHIFT_CAPTURE_STORE", "EVALSHIFT_SINK", + "EVALSHIFT_CAPTURE_LAYOUT", ): monkeypatch.delenv(var, raising=False) reset_config() yield reset_config() + _layout._guard_layout.cache_clear() @pytest.fixture @@ -52,6 +65,7 @@ def _read(suite: str) -> list[dict[str, Any]]: suite_dir = tmp_path / "captures" / suite if not suite_dir.exists(): return [] - return [json.loads(p.read_text(encoding="utf-8")) for p in sorted(suite_dir.glob("*.json"))] + paths = sorted(suite_dir.glob("*.json")) + sorted(suite_dir.glob("conv_*/*.json")) + return [json.loads(p.read_text(encoding="utf-8")) for p in paths] return _read diff --git a/tests/test_capture_sampling.py b/tests/test_capture_sampling.py index 7c7d718..782385d 100644 --- a/tests/test_capture_sampling.py +++ b/tests/test_capture_sampling.py @@ -128,3 +128,68 @@ def handle() -> str: handle() assert len(read_captures("samp")) == 1 + + +def test_a_conversation_is_sampled_whole( + capturing: Path, read_captures: CaptureReader, monkeypatch: pytest.MonkeyPatch +) -> None: + # rate 0.55: "planner:session-1" draws 0.4956 (kept), "c1" draws 0.5988 (dropped) -- on every + # turn, whatever random.random says. + configure(sample_rate=0.55) + draws = iter([0.0, 0.99, 0.0, 0.99, 0.0, 0.99]) + monkeypatch.setattr("random.random", lambda: next(draws)) + for conversation in ("planner:session-1", "c1"): + for turn in range(3): + with capture.agent_session( + suite="s", + redact=False, + agent_input=f"{conversation} {turn}", + conversation_id=conversation, + turn_index=turn, + tools=[], + ): + record_model_call(model_id="m", tools=[]) + kept = read_captures("s") + assert {c["conversation_id"] for c in kept} == {"planner:session-1"} + assert sorted(c["turn_index"] for c in kept) == [0, 1, 2] + + +async def test_an_async_conversation_is_sampled_whole( + capturing: Path, read_captures: CaptureReader, monkeypatch: pytest.MonkeyPatch +) -> None: + # The async twin of the test above: same rate, same ids, same random draws. + configure(sample_rate=0.55) + draws = iter([0.0, 0.99, 0.0, 0.99, 0.0, 0.99]) + monkeypatch.setattr("random.random", lambda: next(draws)) + for conversation in ("planner:session-1", "c1"): + for turn in range(3): + async with capture.agent_session_async( + suite="s", + redact=False, + agent_input=f"{conversation} {turn}", + conversation_id=conversation, + turn_index=turn, + tools=[], + ): + record_model_call(model_id="m", tools=[]) + kept = read_captures("s") + assert {c["conversation_id"] for c in kept} == {"planner:session-1"} + assert sorted(c["turn_index"] for c in kept) == [0, 1, 2] + + +def test_the_decorators_static_conversation_id_samples_at_random( + capturing: Path, read_captures: CaptureReader, monkeypatch: pytest.MonkeyPatch +) -> None: + # The decorator's id is fixed at decoration time: hashing it would keep every run or none. + configure(sample_rate=0.55) + draws = iter([0.0, 0.99, 0.0, 0.99]) + monkeypatch.setattr("random.random", lambda: next(draws)) + + @capture.agent(suite="s", redact=False, tools=[], conversation_id="c1") + def run(question: str) -> str: + record_model_call(model_id="m", tools=[]) + return question + + for i in range(4): + run(f"q{i}") + assert len(read_captures("s")) == 2 # the two runs whose random draw was 0.0 diff --git a/tests/test_conversation_layout.py b/tests/test_conversation_layout.py new file mode 100644 index 0000000..cb48470 --- /dev/null +++ b/tests/test_conversation_layout.py @@ -0,0 +1,354 @@ +"""Conversation folders: where a capture lands, on disk and in a store (0.7.0).""" + +from __future__ import annotations + +import importlib.metadata +import logging +from collections.abc import Callable +from pathlib import Path + +import pytest + +from evalshift import capture, configure, record_model_call +from evalshift.capture.span import SpanTree +from evalshift.sinks import _layout +from evalshift.sinks._layout import ( + _cli_version, + conversation_dir, + is_conversation_dir, + resolve_layout, + writing_flat, +) +from evalshift.sinks.file import FileSink +from evalshift.sinks.object_store import ObjectStoreSink +from evalshift.stores.memory import MemoryStore +from evalshift.trace.models import CaptureEnvelope +from evalshift.trace.serialize import build_capture +from tests.conftest import CaptureReader + + +def _envelope(conversation_id: str | None = "c1") -> CaptureEnvelope: + return build_capture( + SpanTree(), + suite="support_agent", + agent_input="hi", + capture_id="cap_abc", + conversation_id=conversation_id, + turn_index=0 if conversation_id is not None else None, + ) + + +@pytest.mark.parametrize( + ("conversation_id", "folder"), + [ + # Pinned in evalshift-cli's tests too: both sides read and write these exact names. + ("c1", "conv_d0f631ca1ddba8db"), + ("planner:session-1", "conv_e1961a13f5bd6642"), + ("", "conv_e3b0c44298fc1c14"), + ], +) +def test_conversation_dir_is_a_stable_hash_of_the_id(conversation_id: str, folder: str) -> None: + assert conversation_dir(conversation_id) == folder + assert is_conversation_dir(folder) + + +@pytest.mark.parametrize( + "name", ["nested", "conv_D0F631CA1DDBA8DB", "conv_d0f631ca", "conv_d0f631ca1ddba8db0"] +) +def test_only_the_sdks_own_folder_names_count(name: str) -> None: + assert not is_conversation_dir(name) + + +def test_file_sink_files_a_conversation_turn_in_its_folder(tmp_path: Path) -> None: + path = FileSink(base=tmp_path).write(_envelope()) + folder = tmp_path / "captures" / "support_agent" / "conv_d0f631ca1ddba8db" + assert path == folder / "cap_abc.json" + assert path is not None + assert path.exists() + + +def test_file_sink_keeps_a_standalone_capture_flat(tmp_path: Path) -> None: + path = FileSink(base=tmp_path).write(_envelope(None)) + assert path == tmp_path / "captures" / "support_agent" / "cap_abc.json" + + +def test_object_store_keys_mirror_the_local_layout() -> None: + store = MemoryStore() + sink = ObjectStoreSink(store, background=False) + sink.write(_envelope()) + sink.write(_envelope(None)) + assert set(store.objects) == { + "captures/support_agent/conv_d0f631ca1ddba8db/cap_abc.json", + "captures/support_agent/cap_abc.json", + } + + +def test_flat_layout_env_restores_the_old_layout( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("EVALSHIFT_CAPTURE_LAYOUT", " FLAT ") + assert FileSink(base=tmp_path).write(_envelope()) == ( + tmp_path / "captures" / "support_agent" / "cap_abc.json" + ) + store = MemoryStore() + ObjectStoreSink(store, background=False).write(_envelope()) + assert set(store.objects) == {"captures/support_agent/cap_abc.json"} + + +def test_an_explicit_layout_wins_over_the_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("EVALSHIFT_CAPTURE_LAYOUT", "flat") + path = FileSink(base=tmp_path, layout="conversation").write(_envelope()) + assert path is not None + assert path.parent.name == "conv_d0f631ca1ddba8db" + + +@pytest.mark.parametrize("raw", [None, "", "conversation", "Conversation", "nested", "0"]) +def test_anything_but_flat_means_conversation_without_an_old_cli( + raw: str | None, monkeypatch: pytest.MonkeyPatch +) -> None: + # conftest makes the co-install guard see no evalshift CLI. + if raw is None: + monkeypatch.delenv("EVALSHIFT_CAPTURE_LAYOUT", raising=False) + else: + monkeypatch.setenv("EVALSHIFT_CAPTURE_LAYOUT", raw) + assert resolve_layout() == "conversation" + + +def test_a_session_turn_lands_in_its_conversation_folder( + capturing: Path, read_captures: CaptureReader +) -> None: + for turn in range(2): + with capture.agent_session( + suite="planner", + redact=False, + agent_input=f"turn {turn}", + conversation_id="planner:session-1", + turn_index=turn, + tools=[], + ): + record_model_call(model_id="claude-opus-4-8", tools=[]) + folder = capturing / "captures" / "planner" / "conv_e1961a13f5bd6642" + assert len(list(folder.glob("*.json"))) == 2 + assert sorted(c["turn_index"] for c in read_captures("planner")) == [0, 1] + + +# R10: a decorator's conversation_id is fixed at decoration time, so it names no conversation; +# its captures stay flat (with the id still in the envelope). + + +def test_writing_flat_overrides_the_layout_only_inside_the_block() -> None: + with writing_flat(): + assert resolve_layout("conversation") == "flat" + assert resolve_layout("conversation") == "conversation" + + +def test_decorator_capture_stays_flat_with_its_conversation_id( + capturing: Path, read_captures: CaptureReader +) -> None: + @capture.agent(suite="planner", redact=False, conversation_id="c1", turn_index=0, tools=[]) + def handle(query: str) -> str: + record_model_call(model_id="claude-opus-4-8", tools=[]) + return query + + handle("hi") + suite_dir = capturing / "captures" / "planner" + assert [p.parent for p in suite_dir.rglob("*.json")] == [suite_dir] + [cap] = read_captures("planner") + assert cap["conversation_id"] == "c1" + assert cap["turn_index"] == 0 + + +async def test_async_decorator_capture_stays_flat( + capturing: Path, read_captures: CaptureReader +) -> None: + @capture.agent(suite="planner", redact=False, conversation_id="c1", turn_index=0, tools=[]) + async def handle(query: str) -> str: + record_model_call(model_id="claude-opus-4-8", tools=[]) + return query + + await handle("hi") + suite_dir = capturing / "captures" / "planner" + assert [p.parent for p in suite_dir.rglob("*.json")] == [suite_dir] + [cap] = read_captures("planner") + assert cap["conversation_id"] == "c1" + + +def test_decorator_capture_key_is_flat_in_a_store(capturing: Path) -> None: + store = MemoryStore() + configure(sink=ObjectStoreSink(store, background=False)) + + @capture.agent(suite="planner", redact=False, conversation_id="c1", turn_index=0, tools=[]) + def handle() -> str: + record_model_call(model_id="claude-opus-4-8", tools=[]) + return "ok" + + handle() + [key] = [k for k in store.objects if k.startswith("captures/")] + assert key.startswith("captures/planner/cap_") + assert key.count("/") == 2 + + +def test_a_session_after_a_decorator_write_still_gets_its_folder( + capturing: Path, read_captures: CaptureReader +) -> None: + @capture.agent(suite="planner", redact=False, conversation_id="c1", turn_index=0, tools=[]) + def handle() -> str: + record_model_call(model_id="claude-opus-4-8", tools=[]) + return "ok" + + handle() + with capture.agent_session( + suite="planner", + redact=False, + agent_input="turn 1", + conversation_id="c1", + turn_index=1, + tools=[], + ): + record_model_call(model_id="claude-opus-4-8", tools=[]) + suite_dir = capturing / "captures" / "planner" + assert len(list(suite_dir.glob("*.json"))) == 1 + assert len(list((suite_dir / "conv_d0f631ca1ddba8db").glob("*.json"))) == 1 + assert sorted(c["turn_index"] for c in read_captures("planner")) == [0, 1] + + +def test_a_sink_can_be_told_to_write_flat(tmp_path: Path) -> None: + flat = tmp_path / "captures" / "support_agent" / "cap_abc.json" + assert FileSink(base=tmp_path, layout="flat").write(_envelope()) == flat + store = MemoryStore() + ObjectStoreSink(store, background=False, layout="flat").write(_envelope()) + assert set(store.objects) == {"captures/support_agent/cap_abc.json"} + + +def test_an_explicit_store_layout_wins_over_the_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("EVALSHIFT_CAPTURE_LAYOUT", "flat") + store = MemoryStore() + ObjectStoreSink(store, background=False, layout="conversation").write(_envelope()) + assert set(store.objects) == {"captures/support_agent/conv_d0f631ca1ddba8db/cap_abc.json"} + + +# R11: an evalshift CLI older than 1.6 in the same environment cannot read conversation folders, +# so the SDK writes flat there unless the user opts in. + + +def _cli(monkeypatch: pytest.MonkeyPatch, version: str | Callable[[], str]) -> None: + lookup = version if callable(version) else (lambda: version) + monkeypatch.setattr(_layout, "_cli_version", lookup) + _layout._guard_layout.cache_clear() + + +def test_an_old_cli_alongside_means_flat_with_one_warning( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + _cli(monkeypatch, "1.5.0") + with caplog.at_level(logging.WARNING, logger="evalshift"): + first = FileSink(base=tmp_path).write(_envelope()) + second = FileSink(base=tmp_path).write(_envelope()) + assert first == second == tmp_path / "captures" / "support_agent" / "cap_abc.json" + [record] = [r for r in caplog.records if r.levelno == logging.WARNING] + assert record.name == "evalshift" + assert "1.5.0" in record.getMessage() + assert "EVALSHIFT_CAPTURE_LAYOUT=conversation" in record.getMessage() + + +@pytest.mark.parametrize("version", ["1.6.0", "1.6.0rc1", "1.6.0.dev0", "2.0", "10.1.3"]) +def test_a_new_enough_cli_means_conversation(version: str, monkeypatch: pytest.MonkeyPatch) -> None: + _cli(monkeypatch, version) + assert resolve_layout() == "conversation" + + +@pytest.mark.parametrize("version", ["0.9.9", "1.5.0", "1.5.99", "1.0"]) +def test_an_older_cli_means_flat(version: str, monkeypatch: pytest.MonkeyPatch) -> None: + _cli(monkeypatch, version) + assert resolve_layout() == "flat" + + +def _raise(exc: BaseException) -> Callable[[], str]: + def lookup() -> str: + raise exc + + return lookup + + +@pytest.mark.parametrize( + "version", + [ + "", + "garbage", + "v1.5.0", + _raise(importlib.metadata.PackageNotFoundError("evalshift")), + _raise(RuntimeError("metadata exploded")), + ], +) +def test_no_readable_cli_version_means_conversation( + version: str | Callable[[], str], monkeypatch: pytest.MonkeyPatch +) -> None: + _cli(monkeypatch, version) + assert resolve_layout() == "conversation" + + +def test_the_guard_looks_up_the_cli_distribution(monkeypatch: pytest.MonkeyPatch) -> None: + seen: list[str] = [] + + def version(name: str) -> str: + seen.append(name) + return "1.5.0" + + monkeypatch.setattr(importlib.metadata, "version", version) + _cli(monkeypatch, _cli_version) # the real lookup, not conftest's "not installed" + assert resolve_layout() == "flat" + assert seen == ["evalshift"] + + +def test_the_guard_is_computed_once_per_process(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[None] = [] + + def lookup() -> str: + calls.append(None) + return "1.5.0" + + _cli(monkeypatch, lookup) + assert [resolve_layout() for _ in range(3)] == ["flat"] * 3 + assert len(calls) == 1 + + +@pytest.mark.parametrize("raw", ["conversation", " Conversation "]) +def test_the_env_opts_in_to_conversation_over_an_old_cli( + raw: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _cli(monkeypatch, "1.5.0") + monkeypatch.setenv("EVALSHIFT_CAPTURE_LAYOUT", raw) + path = FileSink(base=tmp_path).write(_envelope()) + assert path is not None + assert path.parent.name == "conv_d0f631ca1ddba8db" + + +@pytest.mark.parametrize("raw", ["", "nested"]) +def test_an_unset_or_unknown_env_leaves_the_guard_in_charge( + raw: str, monkeypatch: pytest.MonkeyPatch +) -> None: + _cli(monkeypatch, "1.5.0") + monkeypatch.setenv("EVALSHIFT_CAPTURE_LAYOUT", raw) + assert resolve_layout() == "flat" + + +def test_an_explicit_layout_wins_over_an_old_cli( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _cli(monkeypatch, "1.5.0") + path = FileSink(base=tmp_path, layout="conversation").write(_envelope()) + assert path is not None + assert path.parent.name == "conv_d0f631ca1ddba8db" + store = MemoryStore() + ObjectStoreSink(store, background=False, layout="conversation").write(_envelope()) + assert set(store.objects) == {"captures/support_agent/conv_d0f631ca1ddba8db/cap_abc.json"} + + +def test_writing_flat_wins_over_everything(monkeypatch: pytest.MonkeyPatch) -> None: + _cli(monkeypatch, "2.0.0") + monkeypatch.setenv("EVALSHIFT_CAPTURE_LAYOUT", "conversation") + with writing_flat(): + assert resolve_layout("conversation") == "flat" + assert resolve_layout() == "flat" diff --git a/tests/test_hygiene_gc.py b/tests/test_hygiene_gc.py index 3587038..0517014 100644 --- a/tests/test_hygiene_gc.py +++ b/tests/test_hygiene_gc.py @@ -3,7 +3,10 @@ from __future__ import annotations import os +import random +import shutil from pathlib import Path +from typing import Any import pytest @@ -118,3 +121,138 @@ def flaky_unlink(self: Path, missing_ok: bool = False) -> None: removed = gc.evict(tmp_path, max_captures=1, capture_ttl_seconds=None, now_ts=1000.0) assert removed == [b] # b removed despite a failing assert a.exists() # the failing one survives, no exception raised + + +CONV_A = "conv_" + "a" * 16 +CONV_B = "conv_" + "b" * 16 +CONV_C = "conv_" + "c" * 16 + + +def _turns(directory: Path, folder: str, *mtimes: float) -> list[Path]: + conv = directory / folder + conv.mkdir(parents=True, exist_ok=True) + return [_make(conv, f"cap_{folder[-2:]}_{i}.json", mtime=m) for i, m in enumerate(mtimes)] + + +def test_the_boundary_conversation_keeps_its_newest_turns_and_older_ones_go_whole( + tmp_path: Path, +) -> None: + oldest = _turns(tmp_path, CONV_C, 50.0, 60.0) + boundary = _turns(tmp_path, CONV_A, 100.0, 101.0, 102.0) + new = _turns(tmp_path, CONV_B, 200.0, 201.0) + removed = gc.evict(tmp_path, max_captures=4, capture_ttl_seconds=None, now_ts=1000.0) + # B is kept whole (2), A keeps the 2 newest turns that still fit, C goes whole. + assert set(removed) == {*oldest, boundary[0]} + assert not (tmp_path / CONV_C).exists() + assert boundary[1].exists() and boundary[2].exists() + assert all(p.exists() for p in new) + + +def test_a_conversation_larger_than_the_cap_keeps_only_its_newest_turns(tmp_path: Path) -> None: + # A conversation that never ends (a static decorator id, a user id): disk stays bounded. + live = _turns(tmp_path, CONV_A, 300.0, 301.0, 302.0) + older = _make(tmp_path, "cap_flat.json", mtime=100.0) + removed = gc.evict(tmp_path, max_captures=2, capture_ttl_seconds=None, now_ts=1000.0) + assert removed == [older, live[0]] + assert live[1].exists() and live[2].exists() + + +def test_ttl_evicts_every_old_file_so_an_idle_conversation_goes_whole(tmp_path: Path) -> None: + idle = _turns(tmp_path, CONV_A, 100.0, 110.0) + active = _turns(tmp_path, CONV_B, 100.0, 990.0) # first turn old, latest recent + removed = gc.evict(tmp_path, max_captures=None, capture_ttl_seconds=500.0, now_ts=1000.0) + assert set(removed) == {*idle, active[0]} + assert not (tmp_path / CONV_A).exists() + assert active[1].exists() + + +def test_flat_and_conversation_units_share_one_cap_oldest_first(tmp_path: Path) -> None: + flat_old = _make(tmp_path, "cap_old.json", mtime=50.0) + conv = _turns(tmp_path, CONV_A, 100.0, 150.0) + flat_new = _make(tmp_path, "cap_new.json", mtime=200.0) + removed = gc.evict(tmp_path, max_captures=3, capture_ttl_seconds=None, now_ts=1000.0) + assert removed == [flat_old] + assert flat_new.exists() and all(p.exists() for p in conv) + + +def test_a_folder_that_is_not_a_conversation_is_never_touched(tmp_path: Path) -> None: + other = tmp_path / "nested" + other.mkdir() + keep = _make(other, "cap_x.json", mtime=1.0) + _make(tmp_path, "cap_new.json", mtime=200.0) + gc.evict(tmp_path, max_captures=1, capture_ttl_seconds=1.0, now_ts=1000.0) + assert keep.exists() + + +def test_the_count_is_of_what_the_ttl_leaves(tmp_path: Path) -> None: + # A's turn at 100 is past the TTL; counting it would push A out beside the flat capture. + a = _turns(tmp_path, CONV_A, 100.0, 900.0) + flat = _make(tmp_path, "cap_flat.json", mtime=950.0) + removed = gc.evict(tmp_path, max_captures=2, capture_ttl_seconds=500.0, now_ts=1000.0) + assert removed == [a[0]] + assert a[1].exists() and flat.exists() + + +def test_one_newer_capture_does_not_evict_a_conversation_at_the_cap(tmp_path: Path) -> None: + turns = _turns(tmp_path, CONV_A, *(100.0 + i for i in range(5))) + flat = _make(tmp_path, "cap_flat.json", mtime=200.0) + removed = gc.evict(tmp_path, max_captures=5, capture_ttl_seconds=None, now_ts=1000.0) + assert removed == [turns[0]] # the conversation keeps its newest cap - 1 turns + assert flat.exists() and all(p.exists() for p in turns[1:]) + + +def _scandir_failing_for(name: str, *, vanish: bool = False) -> Any: + real = os.scandir + + def fake(path: Any = ".") -> Any: + if isinstance(path, (str, os.PathLike)) and Path(path).name == name: + if not vanish: + raise PermissionError(f"denied: {path}") + shutil.rmtree(path) # gone between the suite scan and this one + return real(path) + + return fake + + +@pytest.mark.parametrize("vanish", [False, True]) +def test_an_unreadable_or_vanished_folder_does_not_stop_the_rest_of_the_gc( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, vanish: bool +) -> None: + blocked = _turns(tmp_path, CONV_B, 10.0, 20.0) # oldest; skipped, never counted + old = _make(tmp_path, "cap_old.json", mtime=100.0) + conv = _turns(tmp_path, CONV_A, 150.0, 160.0) + new = _make(tmp_path, "cap_new.json", mtime=200.0) + monkeypatch.setattr(os, "scandir", _scandir_failing_for(CONV_B, vanish=vanish)) + removed = gc.evict(tmp_path, max_captures=2, capture_ttl_seconds=None, now_ts=1000.0) + assert removed == [old, conv[0]] + assert new.exists() and conv[1].exists() + assert all(p.exists() for p in blocked) is not vanish + + +def _keep_newest_after_ttl( + files: list[tuple[float, Path]], cap: int | None, ttl: float | None, now: float +) -> set[Path]: + alive = sorted(f for f in files if ttl is None or now - f[0] <= ttl) + return {path for _, path in (alive if cap is None else alive[max(len(alive) - cap, 0) :])} + + +def test_flat_only_suites_keep_the_newest_n_after_the_ttl(tmp_path: Path) -> None: + rng = random.Random(20261021) + for case in range(300): + directory = tmp_path / f"case_{case}" + directory.mkdir() + # Integer mtimes from a small range force ties, so the tie order is checked too. + files = [ + (m, _make(directory, f"cap_{i:03d}.json", mtime=m)) + for i, m in enumerate(float(rng.randint(0, 50)) for _ in range(rng.randint(0, 25))) + ] + cap = rng.choice([None, *range(1, 30)]) + ttl = rng.choice([None, float(rng.randint(0, 60))]) + if cap is None and ttl is None: + cap = rng.randint(1, 30) + now = 50.0 + expected = _keep_newest_after_ttl(files, cap, ttl, now) + removed = gc.evict(directory, max_captures=cap, capture_ttl_seconds=ttl, now_ts=now) + survivors = set(directory.glob("*.json")) + assert survivors == expected, (case, cap, ttl) + assert set(removed) == {p for _, p in files} - expected, (case, cap, ttl) diff --git a/tests/test_hygiene_sample.py b/tests/test_hygiene_sample.py index 4639ecd..26e1f2d 100644 --- a/tests/test_hygiene_sample.py +++ b/tests/test_hygiene_sample.py @@ -2,7 +2,9 @@ from __future__ import annotations -from evalshift.hygiene.sample import should_capture +import pytest + +from evalshift.hygiene.sample import conversation_draw, should_capture def test_none_rate_captures_everything() -> None: @@ -27,3 +29,22 @@ def test_midrate_drops_when_draw_at_or_above_rate() -> None: assert should_capture(0.5, rng=lambda: 0.6) is False # Boundary: draw == rate drops (strict <), keeping the 0.0 case consistent. assert should_capture(0.5, rng=lambda: 0.5) is False + + +@pytest.mark.parametrize( + ("key", "draw"), + [("c1", 0.5987745500564885), ("planner:session-1", 0.49559010417828814)], +) +def test_a_conversations_draw_is_a_stable_hash(key: str, draw: float) -> None: + assert conversation_draw(key) == draw + + +def test_a_conversation_key_ignores_the_random_draw() -> None: + assert should_capture(0.59, rng=lambda: 0.0, key="c1") is False # 0.5988 >= 0.59 + assert should_capture(0.60, rng=lambda: 0.99, key="c1") is True # 0.5988 < 0.60 + + +def test_rate_bounds_still_win_over_a_key() -> None: + assert should_capture(None, key="c1") is True + assert should_capture(1.0, key="c1") is True + assert should_capture(0.0, key="c1") is False diff --git a/tests/test_hygiene_sink.py b/tests/test_hygiene_sink.py index 4fd9c69..9cc198d 100644 --- a/tests/test_hygiene_sink.py +++ b/tests/test_hygiene_sink.py @@ -90,3 +90,24 @@ def boom(*_a: object, **_k: object) -> list[Path]: path = sink.write(_env("a", "cap_1")) assert path is not None assert path.exists() # capture still written despite GC blowing up + + +def test_gc_caps_the_suite_across_conversation_folders(tmp_path: Path) -> None: + sink = HygieneSink(FileSink(base=tmp_path), dedup=False, max_captures=2, capture_ttl=None) + for i, conversation in enumerate(("c1", "c2", "c3")): + path = sink.write( + build_capture( + SpanTree(), + suite="s", + agent_input=f"q{i}", + capture_id=f"cap_{i}", + conversation_id=conversation, + turn_index=0, + ) + ) + assert path is not None + # Deterministic mtimes, oldest first: the next write's GC sees this one as older. + os.utime(path, (100.0 + i, 100.0 + i)) + suite_dir = tmp_path / "captures" / "s" + assert sorted(p.name for p in suite_dir.glob("conv_*/*.json")) == ["cap_1.json", "cap_2.json"] + assert len(list(suite_dir.iterdir())) == 2 # the evicted conversation's folder is gone