diff --git a/ai-company-brain/specs/observability_e2.md b/ai-company-brain/specs/observability_e2.md index 32343129c..f34e571be 100644 --- a/ai-company-brain/specs/observability_e2.md +++ b/ai-company-brain/specs/observability_e2.md @@ -4,10 +4,23 @@ > **shipped** (2026-07-09). Phase 6 + 6.1–6.6 (cross-app cost, per-agent > correlation, access fix + durable history, office UX) and 6.8 (real Pixel Lab > sprites + Avatar Studio) **shipped** (2026-07-09/10). E2 C+ → A. -> **Open: §7 (WS-6) — attribution stamp + durable cost table + the dormant -> Langfuse/OTel half.** Deep tracing is still absent (BO-5). +> **§7 (WS-6): WS-6a + WS-6c shipped 2026-08-02** (incl. repair round 1) — +> decision D1's attribution stamp now exists as a substrate: `instance` +> completes the run context's four-tuple and `_emit_usage` forwards it, so every +> in-run model activation is attributed by (run_id, member, agent, instance) +> with no call-site changes. It reaches two readers: the live presence key +> (`/observability/active` + `/roster`, via `activity.refresh_run_presence`) +> and the daily cost rollup (`cost_summary().by_instance`). Two asymmetries are +> recorded, not hidden: the `phase="start"` **stream** event predates the +> partition, and a **delegated sub-run inherits its caller's** partition. +> **Still open: WS-6b** (carry the tuple to the v1_compat choke point), +> **WS-6d** (durable `llm_call` cost table), **WS-6e** (`agent_run` token +> columns), and the owner-gated Langfuse/OTel half (WS-6f–i). Nothing durable is +> written yet — the stamp lands on logs, the Redis feed and the Redis day-hash +> only. +> Deep tracing is still absent (BO-5). > **Verified against code on 2026-08-01** (paths, `agent_run` token columns, -> `bind_run_context` key set, uv.lock telemetry deps). +> uv.lock telemetry deps); `bind_run_context` key set re-verified 2026-08-02. > **Module:** E2 (core_module_map.md). > **Goal (user request):** log every agent/model interaction so an engineer can > debug "error X happened with agent Y" after the fact (Phases 1–4); AND give @@ -63,12 +76,15 @@ reachability. ## What shipped — Phases 1+2 ### Phase 1 — Correlated, JSON-able logs (`packages/acb_common/acb_common/_log.py`) -- `bind_run_context(run_id, thread_id, agent, user)` / `clear_run_context()` / - `get_run_context()` bind the run fields into structlog contextvars. The - executor binds them at the run boundary in `run_agent_stream` (and clears in - the `finally`), so **every log line the run emits — across all tiers and - injected tools on that context — automatically carries them**. Verified: - `agent.step` and `acb_llm.usage` (which passes no ids) both come out tagged. +- `bind_run_context(run_id, thread_id, agent, user, source, instance)` / + `clear_run_context()` / `get_run_context()` bind the run fields into structlog + contextvars. The executor binds them at the run boundary in `run_agent_stream` + (and clears in the `finally`), so **every log line the run emits — across all + tiers and injected tools on that context — automatically carries them**. + Verified: `agent.step` and `acb_llm.usage` (which passes no ids) both come out + tagged. `instance` (§7 WS-6a) is bound by a second, additive call once the + agent config is loaded — see `executor._bind_run_instance`; binds are additive + by design, so a later call never disturbs an earlier one. - `configure_logging(level, json_logs=?)` + `LOG_FORMAT` env: `LOG_FORMAT=json` → `JSONRenderer` (one JSON object per line, greppable / aggregator-ready); default stays the colored console renderer for local dev. **Prod turns this @@ -495,20 +511,23 @@ optional cleanups: (a) the shadowed duplicate `/v1/chat/completions` in `main.py > 429s, degrade-to-read-only) is **WS-16** per D2/D8 — do not build it here even > though `routes/apps/runtime.py` shows how. WS-16 is gated on this slice. -### Where we actually are (re-verified against the tree 2026-08-01) +### Where we actually are (table re-verified 2026-08-02, after WS-6a/6c) | D1 field | State today | Anchor | |---|---|---| -| `run_id` | Bound in the run context; **dropped** before v1_compat | `packages/acb_common/acb_common/_log.py:68` `_RUN_CONTEXT_KEYS` | -| `member_email` | Bound as `user` in the run context; **dropped** before v1_compat | same | -| `agent` | ✅ carried, via `X-CC-Agent` | `routes/v1_compat.py:425` | -| `instance` | ❌ **does not exist anywhere** in the observability path | `grep -rn instance_id apps/ packages/` → no hits | - -`_RUN_CONTEXT_KEYS` is `("run_id", "thread_id", "agent", "user", "source")` — no -`instance`. v1_compat reads only `x-cc-agent` and `x-cc-source` -(`routes/v1_compat.py:425-426`) and forwards them into `_emit_usage` -(`:573` streaming, `:594` non-streaming); run_id and member are lost because -v1_compat is a bare HTTP request with no inherited context. +| `run_id` | Bound in the run context, inherited by model events; **still dropped** before v1_compat | `packages/acb_common/acb_common/_log.py` `_RUN_CONTEXT_KEYS` | +| `member_email` | Bound as `user` in the run context, inherited; **still dropped** before v1_compat | same | +| `agent` | ✅ carried, via `X-CC-Agent`, and inherited in-run | `routes/v1_compat.py:425` | +| `instance` | ✅ **bound in the run context (WS-6a)**, inherited onto model activations (WS-6c), patched onto the live presence key (so `/observability/active` + `/roster` carry it) and folded into the daily cost rollup as `by_instance`; **not** carried to v1_compat | `_log.py` `_RUN_CONTEXT_KEYS`, `executor._bind_run_instance`, `activity.refresh_run_presence`, `activity._record_cost` | + +`_RUN_CONTEXT_KEYS` is now +`("run_id", "thread_id", "agent", "user", "source", "instance")`. +**The remaining hole is v1_compat, and it is WS-6b's** — it reads only +`x-cc-agent` and `x-cc-source` (`routes/v1_compat.py:425-426`) and forwards +them into `_emit_usage` (`:573` streaming, `:594` non-streaming); run_id, +member and instance are lost because v1_compat is a bare HTTP request with no +inherited context. `_emit_usage` now *accepts* all three (WS-6c) — nothing +passes them yet. **Do not invent a second instance key.** The vocabulary already exists and is authoritative: `_resolve_agent_instance()` @@ -525,34 +544,138 @@ gone on a Redis flush) plus the opt-in `audit_event` row behind (:93)), already read by `routes/apps/runtime.py::_month_ai_usage` (:107). Model the new table on it. -### Slice IN — one agent-safe PR +### Slice IN — the agent-safe items + +The five items below are **AGENT-SAFE**: **a–c** build the stamp, **d** is +where it lands durably, **e** closes the lie the API already tells. -All five items below are **AGENT-SAFE** and belong in one PR: **a–c** build the -stamp, **d** is where it lands durably, **e** closes the lie the API already -tells. None of them needs a decision this spec doesn't record. +**Shipped 2026-08-02 (one PR): WS-6a + WS-6c — the substrate.** They were split +out and dispatched alone because they change no wire protocol, no schema and no +auth surface: the run context gains a key and the emitter gains three optional +parameters that nothing passes yet. **WS-6b, WS-6d and WS-6e remain open** — +they are the consumers, and each still needs a decision recorded before it is +dispatchable (6b: where `member_email` may come from, per the identity trap; +6d: how that distinction is spelled in the schema; 6e: the upsert's +token-column semantics). --- -**WS-6a — Bind `instance` into the run context.** AGENT-SAFE -Extend `_RUN_CONTEXT_KEYS` and `bind_run_context(...)` with an `instance` field +**WS-6a — Bind `instance` into the run context.** ✅ **SHIPPED 2026-08-02** +`_RUN_CONTEXT_KEYS` and `bind_run_context(...)` gained an `instance` field (same "only non-empty values are bound" rule; `clear_run_context` stays -symmetric because it unbinds the key tuple). Bind it from +symmetric because it unbinds the key tuple), bound from `_resolve_agent_instance()`'s value for the run. -- ⚠️ **Ordering trap — this is the whole difficulty of the item.** In - `run_agent_stream`, `bind_run_context` fires at `executor.py:2169`, but - `_agent_instance` is not resolved until `executor.py:2322` because - `_resolve_agent_instance` needs `loaded.config`. A single bind at :2169 can - never carry the instance. Either move the bind after the load, or issue a - second `bind_run_context(instance=…)` once resolved. The second bind is the - smaller blast radius (the first bind must stay early so failures *during* - load are still correlated). -- The other resolve site (`executor.py:1728`) needs the same treatment. +- ⚠️ **Ordering trap — this was the whole difficulty of the item.** In + `run_agent_stream` the `bind_run_context` call sits near the top, but + `_agent_instance` is not resolved until well after `load_agent` returns, + because `_resolve_agent_instance` needs `loaded.config`. One bind can never + carry the instance. + **Resolved as: a second, additive bind** — `executor._bind_run_instance()`, + issued at the resolve site. The first bind stays early so failures *during* + load are still correlated, and only one key moves. (`bind_run_context` is + additive by contract, documented in its docstring — that property is now + load-bearing, so don't "optimise" it into a replace.) +- **Known asymmetry #1 — start vs end (deliberate, and its presence-key + consequence is FIXED).** The agent `phase="start"` activity event is + published before the load, so it carries no `instance`; the `phase="end"` + event (still emitted before `clear_run_context()` in the `finally`) inherits + it. That is honest — at start time the partition genuinely is not yet known — + but anything joining *stream* events by run must not assume both ends carry + it. + ⚠️ That asymmetry was not merely cosmetic: `_axadd` writes the presence key + `cc:activity:live:{run_id}` **from the start event's body** + (`activity.py`, the `phase == "start"` branch), so `active_runs()` — and + therefore `GET /observability/active` and `GET /observability/roster`, the + office view — could never carry `instance` for **any** run. + **Fixed by patching the presence snapshot**, not by re-publishing the start + event: `acb_common.activity.refresh_run_presence(run_id, **fields)` merges + fields into the existing key (`SET … XX`, TTL refreshed) and + `executor._bind_run_instance(instance, run_id)` calls it right after the + bind. Rationale: presence is a *snapshot*, so overwriting it is invisible + and idempotent, whereas a second `start` **stream** entry is visible to + every consumer (SSE feed, office) and reads as a second activation. A miss + is a no-op by design — a patch that arrives after the end event must not + resurrect a finished run. Pinned by + `tests/unit/test_activity_bus.py::test_presence_carries_the_instance_after_ + the_late_bind` (asserts through `active_runs()` and that no stream entry is + added), `::test_presence_refresh_never_resurrects_a_finished_run`, and + `tests/unit/test_instance_wiring.py::test_the_late_bind_also_patches_the_ + live_presence_key`. +- **Known asymmetry #2 — delegation inherits the caller's partition.** + `_run_sub_agent_streaming` (`executor.py:500`) neither resolves nor binds an + instance, and it cannot unbind the caller's: `bind_context` is additive, so + clearing would need `unbind_contextvars("instance")` + restore around the + sub-run. Consequence: when a **personal** agent delegates to a **shared** + one, the sub-run's `phase=start/end` events and every `_emit_usage` inside it + carry `instance=u:` — while that same sub-run's `agent_blob` rows + carry `instance=''`, because the sub-run uses `loaded.agent_dir` directly + (`:644`) with no `_resolve_effective_agent_dir` call. + **So the stamp identifies the partition of the run that RESOLVED it, not + necessarily the partition that run's artefacts live in.** Stated where it + can mislead: `_log.py`'s `_RUN_CONTEXT_KEYS` comment, + `_bind_run_instance`'s docstring, `packages/AGENTS.md` and + `apps/services/orchestrator/AGENTS.md` §9. + **Consequence for WS-6d:** the durable `llm_call` row's `instance` is the + *billing/caller* partition, which is the right subject for a per-member cap + but is **not** a foreign key onto `agent_blob.instance` — do not model it as + a join and do not reconcile the two tables on that column. Giving delegation + its own partition is a separate ticket (it needs the unbind/restore, and a + decision about whether a delegated shared agent should even be attributed to + the caller's tenant); it is **not** in WS-6d. +- 🚧 **Scope line (deliberate, do not "finish" it casually).** WS-6a covers + `run_agent_stream` **only**. The 2026-08-01 draft said "the other resolve + site (`executor.py:1728`) needs the same treatment" — it does not. + Before this PR `bind_run_context` had exactly ONE call site in the whole + repo, in `run_agent_stream`; **`run_agent` binds no run context at all**, so + stamping it would mean *adding* correlation to a path that never had it — a + behaviour change (new fields on every log line that path emits, and a new + clear obligation in its `finally`) outside this ticket. It is a real gap: + batch runs are unattributed today, and they were unattributed before WS-6. + It belongs to whoever gives `run_agent` a run boundary. - **Done when:** `get_run_context()` inside a run of an instanced agent returns `instance` equal to `_resolve_agent_instance(config, name, actor)` for that run, and `''` for a shared agent (absent key, not the string `"''"`); `clear_run_context()` leaves no `instance` behind (extend the existing no-leak assertion in `tests/unit/test_observability.py`). + ✅ Met: `tests/unit/test_instance_wiring.py::test_the_run_context_carries_ + the_key_the_executor_resolved` + `::test_a_shared_run_stamps_no_partition` + (executor composition), `tests/unit/test_observability.py::test_instance_is_ + bound_and_readable_like_every_other_run_field`, `::test_shared_agent_binds_ + no_instance_key_at_all`, `::test_a_second_bind_tops_up_instance_without_ + disturbing_the_first`, and the extended `::test_clear_removes_context_no_leak`. +- 🔒 **Twin-tuple drift gate.** `_log._RUN_CONTEXT_KEYS` (what a run *binds*) + and `activity._INHERIT` (what an event *copies* when its emitter omits it) + must be extended together or a new key half-lands — bound onto every log + line but absent from every activity/cost event, or the reverse. Both + AGENTS.md files said so in prose; it is now enforced by + `tests/unit/test_observability.py::test_inherit_and_run_context_keys_match`, + whose failure message names the offending keys and what to do (same style as + `tests/unit/test_skills_registry.py`'s drift gates). +- 📊 **Interim aggregate — `instance` is a cost-rollup dimension.** Before + this, `instance` existed only on raw stream entries, which `STREAM_MAXLEN` + bounds to ~2000 events: "what did alice's personal agents cost today" was + unanswerable until WS-6d. `_record_cost` now folds an + `instance||cost`/`|calls` field into the same `cc:cost:{day}` hash it + already writes for `model`/`source`/`agent`, and `cost_summary()` surfaces + it as `by_instance` — additive, same route (`GET /observability/cost`), same + response shape as the existing `by_agent`. Pinned by + `test_activity_bus.py::test_cost_rollup_folds_the_tenant_partition` (write + side, incl. "a shared run writes no `instance|…` field at all") and the + extended `::test_cost_summary_aggregates_daily_rollups` (read side). + **This does not close WS-6d.** It is the *live* surface only: bounded by + `COST_TTL_SECONDS` (~45 days), lost on a Redis flush, no per-call row, and + it inherits delegation asymmetry #2 above. The durable per-completion record + remains WS-6d's. + ↳ *Exposure, stated:* a `u:` partition key is a member identifier, and + `GET /observability/cost` is open to any AUTHENTICATED caller (Phase 6.3). + That is not a new class of exposure — the same route's sibling feed already + carries `user` (the member email) on every event, and `by_agent` is already + served there — but Q3's "no retention/PII policy" gap now covers this + dimension too, at a 45-day TTL. - **Files:** `packages/acb_common/acb_common/_log.py`, + `packages/acb_common/acb_common/activity.py` + (`refresh_run_presence`, `_record_cost`, `cost_summary`), + `packages/acb_common/acb_common/__init__.py`, `apps/services/orchestrator/orchestrator/executor.py`. **WS-6b — Carry the four-tuple to the v1_compat choke point.** AGENT-SAFE @@ -639,20 +762,49 @@ may come from.** `apps/agents/agent-email-assistant/agents.py`, `apps/agents/agent-whatsapp-assistant/agents.py`. -**WS-6c — Widen `_emit_usage` to accept and forward the stamp.** AGENT-SAFE -`_emit_usage(model, tier, response, source=, agent=)` gains `run_id=`, -`member=`, `instance=`, defaulting to `get_run_context()` so in-run callers +**WS-6c — Widen `_emit_usage` to accept and forward the stamp.** +✅ **SHIPPED 2026-08-02** +`_emit_usage(model, tier, response, source=, agent=)` gained `run_id=`, +`member=`, `instance=`, deferring to `get_run_context()` so in-run callers (`acompletion_with_fallback`, agent runs) need no call-site change and only -v1_compat passes them explicitly. +an out-of-run choke point passes them explicitly. +- ⚠️ **This ticket was half-shipped before it was written — extend, don't + rebuild.** `acb_common/activity.py`'s `_INHERIT` tuple already copied + `agent`/`user`/`thread_id`/`run_id`/`source` from the run context onto any + event whose caller omitted them, so "zero changes at the call site" was + **already true for four of the five fields**. The genuine work was therefore + small: add `instance` to `_INHERIT` (which needs WS-6a to have something to + inherit) and widen the signature so a caller with no run context of its own + can supply the tuple. `member` maps onto the feed's existing `user` field — + no second vocabulary. - **Done when:** an in-run model call is attributed with the full four-tuple with **zero** changes at its call site; the existing activity-event shape gains fields but breaks no consumer (`/observability/activity/*` and the office UI keep rendering). + ✅ Met: `tests/unit/test_llm_usage_telemetry.py::test_in_run_call_is_ + attributed_with_no_call_site_change` asserts all four fields on the event as + it reaches `_axadd` after a call with **no** attribution kwargs; + `::test_existing_source_and_agent_kwargs_are_untouched` pins the pre-existing + caller shape; `::test_shared_agent_run_emits_no_instance_field` and + `test_activity_bus.py::test_build_event_for_a_shared_run_carries_no_instance` + pin that a shared run's event is byte-identical to its pre-WS-6 shape. + Consumers: the gateway serves these events as raw dicts (no response model) + and `workbench/control_plane/src/app/observability/page.tsx:39` types them + with a structural TS `interface` + a `JSON.parse(...) as ActivityEvent` cast, + so an added optional field is inert at runtime. - **Preserve the null-cost contract:** `_compute_cost` (`acb_llm/client.py:480`) returns `None` for an unpriced/stub-registered model and the UI shows "—". **Never coerce that to `0`** — a misleading $0 is worse than an unknown. Pin it. -- **Files:** `packages/acb_llm/acb_llm/client.py`. + ✅ Met and pinned end-to-end: `::test_unpriced_model_publishes_unknown_cost_ + never_zero` — the published event carries no `cost_usd` at all (`None` is + dropped by `_build_event`), so `_record_cost` skips it rather than folding a + fake $0 into the daily rollup. +- **Not done here:** having v1_compat actually *pass* the stamp is **WS-6b** + (header propagation + the identity constraint), which is held. This ticket + builds the socket, not the plug. +- **Files:** `packages/acb_llm/acb_llm/client.py`, + `packages/acb_common/acb_common/activity.py` (`_INHERIT`). **WS-6d — Durable `llm_call` cost table (the deferred durable cost table).** AGENT-SAFE One durable row per completion, modelled on `app_audit`. **R1: find the next @@ -669,6 +821,12 @@ deploy re-applies every migration file on every deploy. trap) — WS-16 will build per-member caps on these rows, and a rollup that cannot tell the two apart is a budget-evasion vector. This is a schema decision, so getting it wrong here is expensive to undo. +- ⚠️ **`instance` is the CALLER's partition, not the artefact's.** Read + delegation asymmetry #2 under WS-6a before designing the column: a delegated + sub-run stamps the caller's `instance` while writing `agent_blob` rows under + `''`. So model this column on `agent_blob`'s *vocabulary*, but do not treat + it as a foreign key onto `agent_blob.instance` and do not build a + reconciliation that joins the two on it. - Index for the rollups the D1 decision names: at least `(at)`, `(member_email, at)`, `(agent, at)`. - The Redis day-rollup stays as-is (it is the *live* surface); this is the @@ -896,7 +1054,26 @@ uv run pytest tests/unit/test_observability.py tests/unit/test_activity_bus.py \ tests/unit/test_llm_usage_telemetry.py tests/unit/test_v1_compat_telemetry.py -q ``` Baseline confirmed 2026-08-01 on a clean tree: **55 passed in 15.14s**. Any WS-6 -PR must keep this green and add to it. +PR must keep this green and add to it. **After WS-6a/6c (2026-08-02) the +baseline is 65 passed in 18.04s** — the +10 are the attribution pins listed in +those tickets' done-whens. **After WS-6a repair round 1 (same day) it is +71 passed in 23.44s** — the +6 are the presence-refresh, twin-tuple drift and +cost-dimension pins. + +The executor half of WS-6a is pinned separately, because `_bind_run_instance` +lives next to `_resolve_agent_instance`: + +``` +uv run pytest tests/unit/test_instance_wiring.py -q +``` +Baseline confirmed 2026-08-02: **18 passed in 15.96s** (15 before WS-6a; 17 +after WS-6a, +1 for the presence-key patch in repair round 1). + +⚠️ `tests/unit/test_run_agent_stream_e2e.py` — the one file that drives the real +`run_agent_stream` — **hangs on this Windows box** (>5 min, no output), like the +full suite. It is therefore NOT the place to pin run-boundary behaviour from +here; test the composition (`_resolve_agent_instance` → `_bind_run_instance` → +`get_run_context`) instead, as WS-6a does. ``` uv run pytest tests/unit/test_observability_access.py \ @@ -925,26 +1102,15 @@ select-list over the **whole repo**, no path narrowing: uv run ruff check . --select F821,F601,F602,F502,F7,B006 ``` -⚠️ **This command is currently RED on `main`, and not because of WS-6.** Verified -2026-08-01 on a clean tree: - -``` -F821 Undefined name `TurnDecision` apps\services\gateway\gateway\routes\agent.py:166:7 -F821 Undefined name `TurnDecision` apps\services\gateway\gateway\routes\agent.py:225:16 -Found 2 errors. -``` +✅ **This command is GREEN on `main`** — re-verified 2026-08-02 on a clean tree: +`All checks passed!` -Both are string annotations (`-> "TurnDecision"` at `:166`, `decision: -"TurnDecision"` at `:225`) whose type is imported **function-locally** at -`agent.py:179` (`from orchestrator.steer import Route, TurnDecision, route_turn`). -They are safe at runtime — a string annotation is never evaluated — but ruff -resolves names lexically and cannot see a name imported inside a different -function body. **This is pre-existing, is not caused by WS-6, and is being fixed -separately.** A WS-6 implementer who sees pr-check's lint job red must confirm -the failure is exactly these two lines and then stop — do not go hunting through -your own diff, and do not "fix" it by widening the import in a WS-6 PR (that is -someone else's ticket and would collide). WS-6's obligation is that the repo-wide -count does not **grow** beyond these two. +*(Corrected 2026-08-02, WS-6a/6c PR. The 2026-08-01 audit recorded this gate as +RED with two pre-existing `F821 Undefined name TurnDecision` errors at +`routes/agent.py:166` / `:225` — string annotations whose type was imported +function-locally. That was fixed separately and the claim is now stale, so it is +removed rather than explained: the gate is green, and **any failure a WS-6 +implementer sees is theirs.**)* Frontend is untouched by this slice; no `next build` gate applies unless the office/cost views change. @@ -1030,9 +1196,10 @@ reads "distributed/OTel tracing **dead** → **BO-5**") plus never a basis for enforcement, plus a done-when pinning a forged header; WS-6d's schema carries the same constraint. (3) Corrected the ruff claim — `pr-check.yml:51` runs the select-list over the **whole repo**, not the - narrowed paths, and it is **currently red on `main`** with 2 pre-existing - `F821 TurnDecision` errors (`routes/agent.py:166`, `:225`) that WS-6 neither - caused nor owns. Also: done-whens written for the owner-gated WS-6f/g/h/i so + narrowed paths. *(Its "red on `main` with 2 pre-existing `F821 TurnDecision` + errors" reading was true on 2026-08-01 and is no longer: the gate is green as + of 2026-08-02 — see Verification.)* Also: done-whens written for the + owner-gated WS-6f/g/h/i so the preamble's claim is true of itself; softened the unsupported "`work_plan.md` §6 enumerates these flags" citation (it does not — that edit belongs to the board's owner); `114_custom_apps.sql:80-91` → `:80-93`; @@ -1040,3 +1207,51 @@ reads "distributed/OTel tracing **dead** → **BO-5**") plus `main.py:1341` anchor with all four `build_orchestrator_agent(with_history= False)` sites (`:366`, `:589`, `:1341`, `:1392`). §7 slice WS-6a–e remains AGENT-SAFE and dispatchable; E2 status otherwise unchanged. +- 2026-08-02 — **§7 WS-6a + WS-6c shipped (code).** The D1 attribution stamp + now exists as a substrate. `_RUN_CONTEXT_KEYS`/`bind_run_context` gained + `instance` (`''` shared binds nothing, so a shared agent's telemetry is + unchanged); `executor._bind_run_instance` tops the run context up right after + `_resolve_agent_instance`, resolving the ordering trap with a **second, + additive bind** rather than moving the early one; `activity._INHERIT` gained + `instance`; `_emit_usage` gained `run_id=`/`member=`/`instance=`, deferring to + the run context. +10 tests (65 in the §7 verification set) plus 2 in + `test_instance_wiring.py`. **Three corrections to the 2026-08-01 audit:** + (1) WS-6c was **half-shipped** — `_INHERIT` already carried four of the five + fields, so "zero call-site changes" was largely pre-existing behaviour, not + new work; (2) WS-6a's "the other resolve site (`:1728`) needs the same + treatment" was **wrong** — `run_agent` binds no run context at all, so that + would add correlation where none existed; recorded as a scope line instead; + (3) the "ruff gate is RED on `main`" note was stale — it is green, and the + claim is removed. **No migration, no schema, no wire-protocol change**: + nothing durable is written and no caller passes the new parameters yet, which + is exactly the WS-6b/6d boundary. +- 2026-08-02 — **§7 WS-6a repair round 1 (code, same PR).** Review of the above + found the stamp reached no *reader*. Four fixes, all inside WS-6a's scope — + still no migration, no schema, no wire-protocol change, nothing durable: + (1) **The presence key could never carry `instance`.** `_axadd` writes + `cc:activity:live:{run_id}` from the `phase="start"` body, which predates the + agent load, so `active_runs()` → `/observability/active` → `/roster` (the + office view) carried no partition for **any** run. New + `acb_common.activity.refresh_run_presence(run_id, **fields)` patches the + existing snapshot (`SET … XX`, TTL refreshed, a miss is a no-op so a + finished run is never resurrected); `_bind_run_instance(instance, run_id)` + calls it. Chosen over re-publishing the start event because a duplicate + `start` **stream** entry is visible to every consumer, while a presence + snapshot overwrite is invisible and idempotent. + (2) **The delegation asymmetry is recorded, not papered over.** + `_run_sub_agent_streaming` resolves no instance and cannot unbind the + caller's, so a delegated sub-run's events carry the caller's partition while + its blobs carry `''`. The three places that asserted the stamp is "the SAME + key the manifest/blob store/state dir use" now say it identifies the + partition of the run that *resolved* it; WS-6a gained asymmetry #2 and WS-6d + gained a "not a foreign key onto `agent_blob`" warning. Fixing delegation + needs `unbind_contextvars` + restore — a separate ticket, deliberately not + done here. + (3) **The twin-tuple invariant is enforced, not prose.** + `test_observability.py::test_inherit_and_run_context_keys_match`. + (4) **The stamp reaches an aggregate.** `_record_cost` folds + `instance||cost`/`|calls` into the existing `cc:cost:{day}` hash and + `cost_summary()` returns `by_instance` — additive, no route change, same + shape as `by_agent`. This is the *live* rollup only (45-day TTL, no per-call + row): **WS-6d is unchanged and still open.** + +8 tests (89 in the §7 verification set incl. `test_instance_wiring.py`). diff --git a/ai-company-brain/work_plan.md b/ai-company-brain/work_plan.md index aa044899b..2c2f39414 100644 --- a/ai-company-brain/work_plan.md +++ b/ai-company-brain/work_plan.md @@ -73,7 +73,7 @@ gap; calendar P3 was found already shipped (with revised roll-over semantics). | WS-3 | **Isolation ladder** (BO-7 / HH-6 / B6 Tier 1→2, `tool_scope` deny, T2 for non-first-party agents) | `permissions_sandbox_b6.md` + `agent_platform_hardening_2026-07.md` Part 1 | 🟢 Tier 1 | Tier 1 container flags partially landed 2026-07-27 (competitive log) — reconcile B6 first. T2 is its own sub-project; required before Agent Workshop opens to non-engineers. **OWNER-GATE:** `AGENT_PERMISSION_MODE=enforce` flip. | | WS-4 | **Event-bus consumer + durable queue** (BO-20) | checklist §BO-20 | 🟢 | `ingestion.worker` confirmed absent. Prereq for WS-11 Slice 4 and multi-channel triggers. | | WS-5 | **CI gates real** (BO-17/BO-18) | checklist §F | 🟡 Docs | Un-gate evals, blocking gitleaks, coverage floor. ~~AGENT-SAFE~~ → **mixed: the highest-value item is a GitHub *settings* change an agent cannot make.** **Audited 2026-08-01 → NO-GO**: §F has zero testable "done when" ("per the existing plan", "a few green PRs", "for foundation packages"), its ratchet-plan anchor points at a path that moved to `specs/archive/` (3 stale citations live *in the workflow files*), and BO-17 reads ☐ while half of it shipped (blocking ruff-correctness + xenon, a frontend tsc/vitest job, gitleaks, per-PR health). **THE MISSING ITEM — why the 2026-08-01 F821 escape happened, in no doc today:** (1) `main` has **no branch protection** (`gh api …/branches/main/protection` → 404) — every "blocking" gate in these YAMLs is decorative; (2) commits pushed straight to main get **zero check-runs** (`15c8933f` had none); (3) `deploy.yml:56-58` lints with the *non-blocking full* `ruff check .`, **not** the `--select F821,…` correctness gate, so deploy went green over a broken tree; (4) PR #318's `pr-check` **failed on that exact F821 and merged anyway**. **Slice when specced (BO-17a "main-guard"):** add a `correctness` job to `deploy.yml` on push-to-main running the `--select` gate, deliberately NOT in the deploy job's `needs:` — loud, not blocking. AGENT-SAFE. **OWNER-GATE:** enabling branch protection / required checks, wiring any gate into `needs:`, removing `skip_tests`; BO-18's purge+rotation is WS-2's, not this row's. Refuted two long-standing beliefs: pr-check **does** cover the frontend, and it **does** run on non-main branches. | -| WS-6 | **Observability wiring + attribution** (BO-5 + decision D1) | `observability_e2.md` §6.7 | 🟡 Docs | OTel exporter (absent from uv.lock), Langfuse wiring, durable cost table, and the (run, member, agent, instance) attribution stamp — prereq for WS-16. **Audited 2026-08-01 → NO-GO on the doc contract, not the work.** The cited "§6.7" is an unnumbered recommendation memo with **zero acceptance criteria**, stale pre-restructure paths (`apps/gateway/…`), no gate labels, and D1's attribution stamp **appears in no spec at all**. Verified drift: `agent_run.{prompt,completion,total}_tokens` are read by `/debug/runs` + `/observability/runs` but **never written** (`run_trace.py::_persist_row` is the only INSERT) — the spec claims they carry cost attribution; they are always NULL. Cost today lives only in Redis day-hashes (45-day TTL) plus opt-in `audit_event`; the durable precedent to generalise is `app_audit` (mig 114), already used for a monthly budget in `routes/apps/runtime.py`. Only 2 of D1's 4 fields reach the choke point (`v1_compat` reads `x-cc-agent`/`x-cc-source`; run_id + member are dropped) and `instance` exists nowhere — reuse `_resolve_agent_instance()` (`executor.py:917`), do not invent a second key. **To unblock:** add a numbered `## 7. WS-6 — open work` to `observability_e2.md` with per-item "done when" + AGENT-SAFE/OWNER-GATE labels, fix the 5 stale paths, correct the token-column claim. Then the dispatchable slice is the **D1 stamp + a durable `llm_call` table** (agent-safe); Langfuse/OTel/`LLM_USAGE_AUDIT` are OWNER-GATE (§6). | +| WS-6 | **Observability wiring + attribution** (BO-5 + decision D1) | `observability_e2.md` **§7** | 🟡 partial | **Docs gate CLEARED** (PR #319 added the numbered §7 with nine lettered tickets WS-6a–i, per-item done-whens and gate labels). **Re-audited 2026-08-02 → GO-NARROWED to WS-6a+WS-6c only.** ✅ **BUILT 2026-08-02, pending review:** D1's attribution stamp exists as a substrate — `instance` joins `_RUN_CONTEXT_KEYS`/`bind_run_context`, resolved once in `run_agent_stream` via a **second additive bind** after `load_agent` (the early bind stays: it is what correlates a failure *during* load; moving it would trade 5 fields for 1), and `_emit_usage` carries the full (run, member, agent, instance) tuple with **zero call-site changes** — it arrives by inheritance via `activity._INHERIT`. Shared agents produce an **absent key, never `''`** (double-guarded + pinned). `refresh_run_presence()` patches `cc:activity:live:{run_id}` after the late bind, so `/observability/active` + `/roster` carry it; interim `by_instance` cost dimension added to the Redis rollup. **Nothing durable is written yet** — logs + Redis feed only. **🔴 WS-6b/6d/6e HELD, still NO-GO:** WS-6b's security amendment names *no workable mechanism* — `bind_run_context` has one call site (`executor.py`), contextvars do not cross the HTTP hop to `v1_compat`, and `agent_run` rows are written at the run *boundary* so a mid-run join finds nothing. **The only mechanism the code supports at request time is the presence key `cc:activity:live:{run_id}`**, which for the orchestrator path carries a server-established `user`; §7 must name it (or name another) before WS-6b dispatches. WS-6e has no token source (`build_run_trace_row` is pure over events+folded) so it sequences *after* WS-6b, not independently; WS-6d additionally waits on the retention/PII answer (Q3). **Two recorded asymmetries** — the `phase="start"` event predates the bind, and **a delegated sub-run inherits the caller's partition** while its blobs key to `''`, so WS-6d must not treat `instance` as a foreign key onto `agent_blob.instance`. **OWNER-GATE:** WS-6f/g/h/i (Langfuse keys, `--profile obs`, `OTEL_EXPORTER_OTLP_ENDPOINT`, `LLM_USAGE_AUDIT`, the MAF telemetry kill switch) — all now listed in §6. | | WS-7 | **Memory activation + search** (BO-21 → BO-22) | checklist §C + `llm_caching_memory.md` | 🔴 | **OWNER-GATE:** flipping `MEM0_ENABLED`/`GRAPHITI_ENABLED` in prod (cost + latent findings in `agent_platform_hardening` Part 5). `acb_search` (BO-22) after. | ### Platform diff --git a/apps/services/orchestrator/AGENTS.md b/apps/services/orchestrator/AGENTS.md index dbb92bcea..10ec99be2 100644 --- a/apps/services/orchestrator/AGENTS.md +++ b/apps/services/orchestrator/AGENTS.md @@ -22,7 +22,8 @@ and streams chat responses as AG-UI events. 6. All agents must go through MAF -- no raw Copilot SDK paths for business execution 7. mutation_runner.py runs inside the Docker sandbox -- uses Copilot SDK directly (by design) 8. workflow_tools.py exposes published Workflows-app workflows to every agent as a three-tool trio (`list_workflows` / `run_workflow` / `get_workflow_run`) — the sibling of app_tools.py, injected through the same `_tool_injection` gated pipeline. Calls go in-process to `gateway.routes.workflows.service` (the same entrypoints the Run button and API use), so concurrency caps, run history, and the approval gates inside a workflow bind agent-triggered runs identically; a run paused at a Human-approval node reports itself as waiting in the approvals inbox — an agent cannot bypass the gate. Spec: ai-company-brain/specs/workflows_app.md F13 -9. _tool_injection.py's static tool collection lives in `_collect_injectable_platform_tools()` (WS-23 S1 — the exact import chain `_inject_agent_tools` always ran, extracted verbatim; injection behavior unchanged). It is the read-only introspection seam the skills catalog (`acb_skills.skill_families` + gateway `GET /integrations/skills`) builds on, and `tests/unit/test_skills_registry.py` drift-fails if an injected tool is missing from the family registry — register any newly injected tool there in exactly one family. WS-23 S2: `_resolve_injected_scope(tool_scope, disabled_families=…)` intersects admin skill toggles (`agent_skill_setting`, loaded once per run by `_load_disabled_skill_families` — same best-effort sync-DB mechanism as app grants/MCP) with the declared scope; the core floor survives anything, the `workflows` toggle is honored at the trio's append site, and NO rows means byte-identical pre-S2 behavior (`tests/unit/test_skill_toggle_enforcement.py` pins all three rules). +9. Run correlation is bound in **two** places in run_agent_stream, deliberately. `bind_run_context(run_id, thread_id, agent, user, source)` fires at the run boundary — BEFORE `load_agent`, so a failure during load is still correlated — and `_bind_run_instance(instance, run_id)` tops up `instance` (the tenant partition from `_resolve_agent_instance`) immediately after the load, because that value needs `loaded.config` and cannot exist earlier. Binds are additive; the single `clear_run_context()` in the `finally` unbinds every key. Do not "simplify" this into one bind: moving the first one late loses correlation on load failures, and dropping the second silently un-attributes every personal/team agent's spend. The same call patches the live presence key via `acb_common.refresh_run_presence` — the `phase="start"` event predates the load, so `/observability/active` and `/observability/roster` would otherwise never show a partition for any run. `_run_sub_agent_streaming` deliberately does **neither**: a delegated sub-run inherits the caller's `instance` (known asymmetry — spec §7 WS-6a) because `bind_context` cannot unbind and a correct fix needs save/restore around the sub-run. Spec: ai-company-brain/specs/observability_e2.md §7 (WS-6a) +10. _tool_injection.py's static tool collection lives in `_collect_injectable_platform_tools()` (WS-23 S1 — the exact import chain `_inject_agent_tools` always ran, extracted verbatim; injection behavior unchanged). It is the read-only introspection seam the skills catalog (`acb_skills.skill_families` + gateway `GET /integrations/skills`) builds on, and `tests/unit/test_skills_registry.py` drift-fails if an injected tool is missing from the family registry — register any newly injected tool there in exactly one family. WS-23 S2: `_resolve_injected_scope(tool_scope, disabled_families=…)` intersects admin skill toggles (`agent_skill_setting`, loaded once per run by `_load_disabled_skill_families` — same best-effort sync-DB mechanism as app grants/MCP) with the declared scope; the core floor survives anything, the `workflows` toggle is honored at the trio's append site, and NO rows means byte-identical pre-S2 behavior (`tests/unit/test_skill_toggle_enforcement.py` pins all three rules). ## Work Guidance diff --git a/apps/services/orchestrator/orchestrator/executor.py b/apps/services/orchestrator/orchestrator/executor.py index aa3988fea..ad52e9716 100644 --- a/apps/services/orchestrator/orchestrator/executor.py +++ b/apps/services/orchestrator/orchestrator/executor.py @@ -937,6 +937,49 @@ def _resolve_agent_instance( return "" +def _bind_run_instance(instance: str, run_id: str = "") -> None: + """Top up this run's correlation context with its tenant partition (WS-6a). + + A SECOND, additive :func:`acb_common.bind_run_context` call rather than a + moved one — deliberately. The run boundary binds run_id/thread_id/agent/ + user/source *before* ``load_agent`` so a failure DURING load is still + correlated, but :func:`_resolve_agent_instance` cannot run until the config + is loaded. Binding only ``instance`` here leaves every earlier field in + place and keeps the blast radius to one key. + + ``''`` (the shared partition) binds nothing — bind_run_context only takes + non-empty values — so a shared agent's context has no ``instance`` key, + which is exactly how the store spells "shared" (migration 136). Never + raises: attribution must not be able to fail a run. + + The live presence key is patched too, given *run_id*. It was written from + the ``phase="start"`` event body, published before the load and therefore + without an instance — so without this refresh ``active_runs()``, + ``/observability/active`` and the office roster could never show a + partition for ANY run. Patching the snapshot beats re-publishing a start + event, which every stream consumer would read as a second activation. + + **Scope — this is the partition of the run that RESOLVED it.** A delegated + sub-run (``_run_sub_agent_streaming``) resolves none and cannot unbind the + caller's, so its events inherit the caller's ``instance`` while its blobs + are written under the shared partition. Known asymmetry, recorded in + ``specs/observability_e2.md`` §7 WS-6a; do not read the stamp as proof + that a given event's blobs live in that partition. + """ + try: + from acb_common import bind_run_context + bind_run_context(instance=instance) + except Exception: + pass + if not instance or not run_id: + return + try: + from acb_common import refresh_run_presence + refresh_run_presence(run_id, instance=instance) + except Exception: + pass + + def _resolve_effective_agent_dir( agent_dir: Path, agent_config: dict[str, Any], @@ -2153,7 +2196,10 @@ async def run_agent_stream( # Bind run_id/thread_id/agent/user into structlog contextvars so EVERY log # line this run emits (across all tiers + injected tools on this context) # carries them — the thing that makes "show me all logs for run X / agent Y" - # possible. Cleared in the finally below. + # possible. Cleared in the finally below. The fifth field of decision D1's + # stamp, `instance`, cannot be resolved yet (it needs loaded.config) and is + # topped up by _bind_run_instance right after the load — this bind stays + # here so a failure DURING the load is still correlated. _corr_source = "chat" _corr_user = "" try: @@ -2322,6 +2368,11 @@ def _respond_input_apply(command: dict[str, Any]) -> bool: _agent_instance = _resolve_agent_instance( loaded.config, agent_name, _corr_user, ) + # Attribution (WS-6a): every log line and every model activation + # from here on carries the partition this run is executing in, and + # the presence snapshot published at start (which predates the + # load, so it has no instance) is patched to match. + _bind_run_instance(_agent_instance, run_id) _effective_ws = _resolve_effective_agent_dir( loaded.agent_dir, loaded.config, session_override=_session_ws, diff --git a/packages/AGENTS.md b/packages/AGENTS.md index d499cb962..a36e1a79e 100644 --- a/packages/AGENTS.md +++ b/packages/AGENTS.md @@ -8,7 +8,7 @@ Reusable Python packages shared across all CommandCenter services. - acb_llm/ -- LiteLLM integration, unified credential store (LLM + integration keys), model routing - acb_memory/ -- Mem0 and Graphiti memory providers. `compartments.py` owns the scope vocabulary (`scope_key`/`scope_kind`: a bare email, `prefs:`, `room:`, `agent:`, `org:global`) and `resolve_clearance` — WHICH compartments a run may read and the ONE it may write (spec: docs/multiplayer/memory-clearance.md §3). Solo resolves to exactly the three scopes and the write target it always had; a shared run swaps the actor's private compartment for the room's and keeps their prefs. The point is that an excluded compartment's scope key is never passed to `search()` — a boundary, not a request in a system prompt. Dependency-free on purpose: the CALLER decides `shared` (the gateway has `resolve_room_access`), which keeps acb_memory below acb_auth and the gateway in the import graph. `session_cache` keys on the clearance fingerprint as well as the thread — without it a thread cached while solo keeps serving the owner's private block after it is shared. - acb_graph/ -- Postgres entity graph (SQLAlchemy sessions) -- acb_common/ -- Shared settings, logging, activity/cost feed, utilities +- acb_common/ -- Shared settings, logging, activity/cost feed, utilities. `_log.py` owns the run-correlation stamp: `_RUN_CONTEXT_KEYS` = `(run_id, thread_id, agent, user, source, instance)` — decision D1's attribution four-tuple plus the thread. Binds are **additive and non-empty-only**, so a caller may top up one field later (the executor resolves `instance` only after the agent config loads) and the shared partition `''` binds nothing rather than an empty value. `activity.py::_INHERIT` copies the same keys onto any event whose emitter omits them — which is why a model call inside a run is attributed with no change at its call site. Extend those two tuples together or attribution silently half-lands (`tests/unit/test_observability.py::test_inherit_and_run_context_keys_match` drift-fails if you don't). `instance` names the partition of the run that RESOLVED it — inheritance means a delegated sub-run carries its caller's key while writing blobs under the shared partition, so the stamp is not a join key onto `agent_blob` without knowing which run emitted it. The presence key (`cc:activity:live:{run_id}`) is written once from the `phase="start"` body, so a field resolved after start reaches `/observability/active` only via `activity.refresh_run_presence`. Spec: ai-company-brain/specs/observability_e2.md §7 - acb_audit/ -- Audit event recording - acb_auth/ -- Authentication, roles, and org access control. Two guard styles coexist: the original coarse `require_role(UserRole.EXECUTIVE)` (unchanged) and `require_permission("feature:whatsapp")`, backed by DB roles + per-user allow/deny overrides (`permissions.py` is pure and testable; `access.py` does the I/O with a 60s cache). `require_authenticated(public=...)` is the app-wide default-deny guard (BO-2 #1) — authentication, not authorization. Two DISTINCT secrets: `GATEWAY_INTERNAL_TOKEN` is service identity (grants everything, never handed to agents) and `LITELLM_MASTER_KEY` is the /v1 API key agents hold, checked only by `require_llm_api_auth`. Spec: ai-company-brain/specs/org_access_control.md diff --git a/packages/acb_common/acb_common/__init__.py b/packages/acb_common/acb_common/__init__.py index 57cb57638..21269ddc7 100644 --- a/packages/acb_common/acb_common/__init__.py +++ b/packages/acb_common/acb_common/__init__.py @@ -12,6 +12,7 @@ publish_activity, read_activity_since, recent_activity, + refresh_run_presence, ) from acb_common.settings import Settings, get_settings @@ -28,4 +29,5 @@ "publish_activity", "read_activity_since", "recent_activity", + "refresh_run_presence", ] diff --git a/packages/acb_common/acb_common/_log.py b/packages/acb_common/acb_common/_log.py index 7c6fc2180..d16c987fc 100644 --- a/packages/acb_common/acb_common/_log.py +++ b/packages/acb_common/acb_common/_log.py @@ -7,9 +7,10 @@ grepped/filtered by field and shipped to an aggregator. Correlation: :func:`bind_run_context` binds ``run_id``/``thread_id``/``agent``/ -``user`` into structlog's contextvars so EVERY log line emitted during a run -carries them automatically — the thing that makes "show me all logs for run X" -possible (E2 observability). Bind at the run boundary, clear in ``finally``. +``user``/``instance`` into structlog's contextvars so EVERY log line emitted +during a run carries them automatically — the thing that makes "show me all logs +for run X" possible (E2 observability). Bind at the run boundary, clear in +``finally``. """ from __future__ import annotations @@ -65,7 +66,21 @@ def get_logger(name: str | None = None) -> structlog.stdlib.BoundLogger: # run gains the same joinable fields (and so bind/clear stay symmetric). # ``source`` is the originating app (chat / email / tasks / …) so the live # activity feed can attribute an activation to the surface that triggered it. -_RUN_CONTEXT_KEYS = ("run_id", "thread_id", "agent", "user", "source") +# ``instance`` is the run's tenant partition, drawn from the SAME vocabulary +# the manifest, the blob store (migration 136) and the state directory use +# (``AgentManifest.instance_key``: ``''`` shared · ``u:`` personal · +# ``t:`` team). It completes decision D1's attribution four-tuple +# (run_id, member, agent, instance) — WS-6a, specs/observability_e2.md §7. +# The value identifies the partition of the run THAT RESOLVED IT; because the +# context is inherited, a delegated sub-run that resolves none carries the +# caller's key while writing blobs under the shared partition (the delegation +# asymmetry recorded in §7 WS-6a). Do not treat the stamp as a join key onto +# ``agent_blob`` without checking which run emitted it. +# ⚠️ Twin tuple: ``activity._INHERIT`` must stay in sync — pinned by +# tests/unit/test_observability.py::test_inherit_and_run_context_keys_match. +_RUN_CONTEXT_KEYS = ( + "run_id", "thread_id", "agent", "user", "source", "instance", +) def bind_run_context( @@ -75,6 +90,7 @@ def bind_run_context( agent: str | None = None, user: str | None = None, source: str | None = None, + instance: str | None = None, ) -> None: """Bind run-correlation fields into structlog contextvars. @@ -82,6 +98,16 @@ def bind_run_context( thread context automatically includes the given fields — so you can filter all log lines for one agent run. Only non-empty values are bound. Pair with :func:`clear_run_context` in a ``finally`` at the run boundary. + + Additive by design: each call binds only the fields it is given, leaving + already-bound ones alone. That is what lets the run boundary bind early + (so failures *during* agent load are still correlated) and top up + ``instance`` once the loaded config makes it resolvable — see + ``orchestrator.executor._bind_run_instance``. + + The shared partition is ``''``, which is falsy, so a shared agent binds no + ``instance`` key at all rather than an empty/quoted value — "absent" is the + wire representation of shared, matching ``agent_blob.instance``. """ fields = { k: v @@ -91,6 +117,7 @@ def bind_run_context( ("agent", agent), ("user", user), ("source", source), + ("instance", instance), ) if v } diff --git a/packages/acb_common/acb_common/activity.py b/packages/acb_common/acb_common/activity.py index 71d8b0d98..dfdb3ea90 100644 --- a/packages/acb_common/acb_common/activity.py +++ b/packages/acb_common/acb_common/activity.py @@ -17,7 +17,11 @@ ``event`` field (same convention as ``stream_relay.push_event``). * **Presence keys** (``cc:activity:live:{run_id}``) track in-flight agent runs (and, identically, in-flight Custom App AI calls — ``kind="app"``) with a - TTL so the "running now" panel survives a crash without leaking. + TTL so the "running now" panel survives a crash without leaking. The key's + body is the ``phase="start"`` event, so a field the run only learns about + *after* start (``instance``) reaches ``active_runs()`` — and therefore + ``/observability/active`` and ``/observability/roster`` — only via + :func:`refresh_run_presence`. This is the live signal only — the durable record stays in ``agent_run`` (migration 50) and the correlated logs. Retention here is bounded by @@ -47,7 +51,12 @@ # Fields copied from the run-correlation context when the caller omits them, so # a model call inside an agent run inherits that run's agent/user/thread/source. -_INHERIT = ("agent", "user", "thread_id", "run_id", "source") +# ``instance`` (WS-6a/6c) rides the same mechanism: because it is bound into +# the run context, a model activation is attributed to the run's tenant +# partition with no change at the emitting call site — completing decision D1's +# (run_id, member, agent, instance) stamp on the live feed. A shared run binds +# no instance, so the field is simply absent, exactly as before. +_INHERIT = ("agent", "user", "thread_id", "run_id", "source", "instance") # ── Shared async client (created lazily on the running loop) ───────────────── @@ -83,12 +92,20 @@ async def _record_cost(r: aioredis.Redis, evt: dict[str, Any]) -> None: """Fold one priced model call into today's rollup hash. Best-effort. One hash per UTC day (``cc:cost:{date}``) with additive fields so a daily - cost breakdown by model / app / agent is a single HGETALL — no migration, - no per-call Postgres write (the deliberate default; LLM_USAGE_AUDIT stays - the opt-in durable path). Fields: + cost breakdown by model / app / agent / tenant partition is a single + HGETALL — no migration, no per-call Postgres write (the deliberate default; + LLM_USAGE_AUDIT stays the opt-in durable path). Fields: total|cost|tokens|calls model||cost|tokens|calls - source||cost · agent||cost + source||cost · agent||cost · instance||cost + + The ``instance`` dimension (WS-6a) is what makes "what did alice's personal + agents cost today" answerable at all: ``instance`` otherwise exists only on + raw stream entries, which are bounded by ``STREAM_MAXLEN``. It is the LIVE + rollup only — bounded by ``COST_TTL_SECONDS`` and lost on a Redis flush; + the durable per-call record is still WS-6d. A shared run carries no + ``instance``, so it folds into the totals and no dimension row, exactly as + before. """ cost = evt.get("cost_usd") if not isinstance(cost, (int, float)): @@ -97,6 +114,7 @@ async def _record_cost(r: aioredis.Redis, evt: dict[str, Any]) -> None: model = str(evt.get("model") or "unknown") source = str(evt.get("source") or "unattributed") agent = str(evt.get("agent") or "") + instance = str(evt.get("instance") or "") key = _cost_key(_today()) pipe = r.pipeline(transaction=False) pipe.hincrbyfloat(key, "total|cost", float(cost)) @@ -110,6 +128,9 @@ async def _record_cost(r: aioredis.Redis, evt: dict[str, Any]) -> None: if agent: pipe.hincrbyfloat(key, f"agent|{agent}|cost", float(cost)) pipe.hincrby(key, f"agent|{agent}|calls", 1) + if instance: + pipe.hincrbyfloat(key, f"instance|{instance}|cost", float(cost)) + pipe.hincrby(key, f"instance|{instance}|calls", 1) pipe.expire(key, COST_TTL_SECONDS) await pipe.execute() @@ -174,8 +195,8 @@ def publish_activity(**fields: Any) -> None: ``phase`` — "start" | "end" (agents + apps; models are single events) ``model``, ``tier`` — model activations ``status``, ``duration_ms`` — agent "end" events - ``agent``, ``user``, ``thread_id``, ``run_id``, ``source`` — inherited - from the current run context when omitted. + ``agent``, ``user``, ``thread_id``, ``run_id``, ``source``, ``instance`` + — inherited from the current run context when omitted. Never raises and never blocks the caller: the Redis write is scheduled onto the running event loop (or a throwaway loop if none is running). A drop is @@ -218,6 +239,93 @@ async def _one_shot() -> None: pass +async def _apply_presence_patch( + r: aioredis.Redis, run_id: str, patch: dict[str, Any], +) -> None: + """Merge *patch* into an EXISTING presence snapshot and refresh its TTL. + + A miss is a deliberate no-op: the key is absent when the run has already + ended (or its start event was dropped), and re-creating it there would + resurrect a finished run in the "running now" panel. ``xx=True`` closes + the same race on the write side, so the worst case is that the patch is + lost — never that presence outlives the run. + """ + key = _live_key(run_id) + raw = await r.get(key) + if not raw: + return + try: + snap = json.loads(raw) + except (json.JSONDecodeError, TypeError): + return + if not isinstance(snap, dict): + return + snap.update(patch) + await r.set( + key, json.dumps(snap, default=str), + ex=LIVE_TTL_SECONDS, xx=True, + ) + + +async def _axpresence(run_id: str, patch: dict[str, Any]) -> None: + await _apply_presence_patch(_get_client(), run_id, patch) + + +def refresh_run_presence(run_id: str, **fields: Any) -> None: + """Patch the live presence snapshot of an in-flight run. Best-effort. + + The presence key is written once, from the ``phase="start"`` event body. + A run that only learns a correlation field *after* it starts — the + executor resolves ``instance`` from ``loaded.config``, which is not + available at the run boundary (WS-6a) — would otherwise leave that field + permanently absent from ``active_runs()``, ``/observability/active`` and + the office roster, for **every** run. + + This patches the snapshot in place instead of re-publishing the start + event: a duplicate ``start`` entry is visible to every stream consumer + (the SSE feed, the office view) and would be read as a second activation. + Presence is a snapshot, so overwriting it is invisible and idempotent. + + Same contract as :func:`publish_activity` — never raises, never blocks, + a drop is acceptable. ``None``/empty values are skipped so a caller can + pass a field unconditionally without writing an empty one. + + Both writes are loop tasks, so a patch issued before its own start event + has been flushed finds no key and is dropped. In practice the agent load + sits between them (many awaits); if it ever loses that race the result is + the pre-fix behaviour — a presence entry without the field — never a + corrupt or resurrected one. + """ + patch = {k: v for k, v in fields.items() if v} + if not run_id or not patch: + return + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + if loop is not None: + try: + task = loop.create_task(_axpresence(run_id, patch)) + task.add_done_callback( + lambda t: None if t.cancelled() else t.exception() + ) + except Exception: # noqa: BLE001 + pass + return + + async def _one_shot() -> None: + r = aioredis.from_url(get_settings().redis_url, decode_responses=True) + try: + await _apply_presence_patch(r, run_id, patch) + finally: + await r.aclose() + + try: + asyncio.run(_one_shot()) + except Exception: # noqa: BLE001 + pass + + def _parse_entry(eid: str, fields: dict[str, str]) -> dict[str, Any] | None: try: evt = json.loads(fields.get("event", "{}")) @@ -271,7 +379,9 @@ async def active_runs() -> list[dict[str, Any]]: """Return the agent runs currently in flight (presence keys not yet expired). Newest-first by start time. Self-healing: a run whose "end" was lost simply - ages out after ``LIVE_TTL_SECONDS``. + ages out after ``LIVE_TTL_SECONDS``. Each entry is the run's start event as + last patched by :func:`refresh_run_presence` — which is how a late-resolved + field like ``instance`` gets here at all. """ r = _get_client() out: list[dict[str, Any]] = [] @@ -313,9 +423,11 @@ def _split_field(field: str) -> tuple[str, str, str]: async def cost_summary(days: int = 7) -> dict[str, Any]: """Aggregate the last *days* daily cost rollups for the /observability cost view. - Returns per-day totals (chronological), plus by-model and by-source rollups - and grand totals. Costs are in USD (best-effort litellm pricing). Never - raises — missing/short history just yields zeroes. + Returns per-day totals (chronological), plus by-model, by-source, by-agent + and by-instance rollups and grand totals. ``by_instance`` is keyed by the + tenant partition (``u:`` / ``t:``); shared runs carry no + partition and appear only in the totals. Costs are in USD (best-effort + litellm pricing). Never raises — missing/short history just yields zeroes. """ days = max(1, min(days, 90)) r = _get_client() @@ -324,6 +436,12 @@ async def cost_summary(days: int = 7) -> dict[str, Any]: by_model: dict[str, dict[str, float]] = {} by_source: dict[str, dict[str, float]] = {} by_agent: dict[str, dict[str, float]] = {} + by_instance: dict[str, dict[str, float]] = {} + # Dimensions whose rollup is just cost+calls keyed by name (unlike `model`, + # which also carries tokens and a per-day breakdown). + flat_dims = { + "source": by_source, "agent": by_agent, "instance": by_instance, + } totals = {"cost": 0.0, "tokens": 0, "calls": 0} for i in range(days): @@ -364,18 +482,15 @@ async def cost_summary(days: int = 7) -> dict[str, Any]: elif metric == "calls": m["calls"] += int(v) dm["calls"] = int(v) - elif dim == "source" and name: - s = by_source.setdefault(name, {"cost": 0.0, "calls": 0}) - if metric == "cost": - s["cost"] += v - elif metric == "calls": - s["calls"] += int(v) - elif dim == "agent" and name: - a = by_agent.setdefault(name, {"cost": 0.0, "calls": 0}) + elif name and dim in flat_dims: + # source / agent / instance accumulate identically — one branch + # instead of three keeps this loop from growing a copy-pasted + # arm per dimension (it is already a C901 hotspot). + e = flat_dims[dim].setdefault(name, {"cost": 0.0, "calls": 0}) if metric == "cost": - a["cost"] += v + e["cost"] += v elif metric == "calls": - a["calls"] += int(v) + e["calls"] += int(v) out_days.append(day_rec) out_days.reverse() # oldest → newest for a left-to-right chart @@ -383,14 +498,16 @@ async def cost_summary(days: int = 7) -> dict[str, Any]: m["cost"] = round(m["cost"], 6) for s in by_source.values(): s["cost"] = round(s["cost"], 6) - for a in by_agent.values(): - a["cost"] = round(a["cost"], 6) + for bucket in flat_dims.values(): + for e in bucket.values(): + e["cost"] = round(e["cost"], 6) totals["cost"] = round(totals["cost"], 6) return { "days": out_days, "by_model": by_model, "by_source": by_source, "by_agent": by_agent, + "by_instance": by_instance, "totals": totals, "window_days": days, } diff --git a/packages/acb_llm/acb_llm/client.py b/packages/acb_llm/acb_llm/client.py index 0d1a121bc..aee77c816 100644 --- a/packages/acb_llm/acb_llm/client.py +++ b/packages/acb_llm/acb_llm/client.py @@ -515,6 +515,8 @@ def _compute_cost(model: str, response: Any, stats: dict[str, int]) -> float | N def _emit_usage( model: str, tier: str, response: Any, *, source: str | None = None, agent: str | None = None, + run_id: str | None = None, member: str | None = None, + instance: str | None = None, ) -> None: """Log per-call token/cache usage + USD cost; optionally persist to audit. @@ -525,6 +527,18 @@ def _emit_usage( header) when the caller knows them; otherwise the run context supplies them. Set ``LLM_USAGE_AUDIT=1`` to also append an ``audit_event`` row per call. Never raises. + + Attribution — decision D1's four-tuple (WS-6c, specs/observability_e2.md + §7). ``run_id``/``member``/``instance`` complete ``agent`` so a completion + can be rolled up per run, per member, per agent and per tenant partition. + **An in-run caller passes none of them**: the run context supplies all four + (``bind_run_context`` binds them; ``activity._INHERIT`` copies them onto the + event), which is why widening this signature costs the ~dozen in-run call + sites zero changes. Only a choke point OUTSIDE a run — the gateway's + ``/v1/chat/completions``, a bare HTTP request that inherits no contextvars — + has anything to pass explicitly. ``member`` is the activity feed's ``user``. + An empty explicit value means "unknown / shared" and defers to the context, + so a shared run stays *absent* rather than gaining an empty field. """ stats = _usage_stats(response) if not stats: @@ -537,12 +551,26 @@ def _emit_usage( _run_ctx = get_run_context() except Exception: _run_ctx = {} - _log.info("acb_llm.usage", model=model, tier=tier, cost_usd=cost, **stats) + # Explicitly-passed attribution, for a caller with no run context of its + # own. Empty values fall through to the run context / are simply omitted — + # never emitted as "". Keyed with the activity feed's own vocabulary + # (member → user) so the log line, the event and the audit row agree. + _stamp = { + k: v for k, v in + (("run_id", run_id), ("user", member), ("instance", instance)) + if v + } + _log.info( + "acb_llm.usage", model=model, tier=tier, cost_usd=cost, + **stats, **_stamp, + ) # Live activity feed (E2): surface every model call on the global bus so the # /observability view shows model activations + cost in real time, across - # every app. Best-effort + non-blocking (never raises); run/agent/user are - # inherited from the run context when this call is inside an agent run, and - # `source` is inherited too unless the caller passes it explicitly. + # every app. Best-effort + non-blocking (never raises); run/agent/user/ + # instance are inherited from the run context when this call is inside an + # agent run, and `source` is inherited too unless the caller passes it + # explicitly. Passing None for the stamp fields is what ARMS that + # inheritance (`activity._INHERIT`) — an explicit value only overrides it. try: from acb_common import publish_activity publish_activity( @@ -553,6 +581,9 @@ def _emit_usage( cost_usd=cost, source=source, agent=agent, + run_id=_stamp.get("run_id"), + user=_stamp.get("user"), + instance=_stamp.get("instance"), ) except Exception: pass @@ -569,7 +600,13 @@ def _persist() -> None: else "system:acb_llm", action="llm_completion", target=f"model:{model}", - payload={"tier": tier, "cost_usd": cost, **stats, **_run_ctx}, + # _stamp last: an explicitly-passed attribution outranks the + # ambient run context (they agree for an in-run call, and only + # an out-of-run caller passes anything at all). + payload={ + "tier": tier, "cost_usd": cost, + **stats, **_run_ctx, **_stamp, + }, )) try: diff --git a/tests/unit/test_activity_bus.py b/tests/unit/test_activity_bus.py index 1eff5c2e8..45a5ca023 100644 --- a/tests/unit/test_activity_bus.py +++ b/tests/unit/test_activity_bus.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import json from acb_common import bind_run_context, clear_run_context, publish_activity from acb_common.activity import _build_event @@ -33,6 +34,32 @@ def test_build_event_inherits_run_context(): assert "ts" in evt # always stamped +def test_build_event_inherits_the_run_instance(): + # WS-6c: `instance` inherits through the SAME _INHERIT mechanism as + # agent/user/run_id, which is what makes a model call inside a personal + # agent's run attributable to that tenant partition with no change at the + # emitting call site. + bind_run_context(run_id="r-i", agent="email-assistant", + user="alice@fracktal.in", instance="u:alice@fracktal.in") + try: + evt = _build_event({"kind": "model", "model": "gpt-x"}) + finally: + clear_run_context() + assert evt["instance"] == "u:alice@fracktal.in" + + +def test_build_event_for_a_shared_run_carries_no_instance(): + # A shared agent binds no instance, so the event simply has no such field — + # the pre-WS-6 shape, byte for byte. Consumers that never heard of + # `instance` see exactly what they saw before. + bind_run_context(run_id="r-s", agent="task-manager", source="chat") + try: + evt = _build_event({"kind": "model", "model": "gpt-x"}) + finally: + clear_run_context() + assert "instance" not in evt + + def test_build_event_explicit_fields_win_over_context(): bind_run_context(agent="sales", source="chat") try: @@ -158,12 +185,26 @@ def __init__(self): async def xadd(self, stream, fields, **_kw): self.xadded.append(fields) - async def set(self, key, value, ex=None): + async def set(self, key, value, ex=None, xx=False): + # `xx=True` means "only if it already exists" — the guard that stops a + # presence patch from resurrecting a run whose end already landed. + if xx and key not in self.store: + return None self.store[key] = value + return True + + async def get(self, key): + return self.store.get(key) async def delete(self, key): self.store.pop(key, None) + async def scan(self, cursor, match=None, count=None): + return 0, list(self.store) + + async def mget(self, keys): + return [self.store.get(k) for k in keys] + def test_app_start_creates_a_presence_key_like_an_agent_run(monkeypatch): # A Custom App AI call (kind="app") is a first-class presence source now, @@ -197,6 +238,87 @@ def test_app_end_clears_the_presence_key(monkeypatch): assert activity._live_key("app-run-2") not in r.store +def test_presence_carries_the_instance_after_the_late_bind(monkeypatch): + # WS-6a: the presence key is written from the phase="start" event, which is + # published BEFORE load_agent and therefore before `instance` exists. Without + # refresh_run_presence the snapshot — and so /observability/active and the + # office roster, which read it verbatim — could never carry a partition for + # ANY run. Patching the snapshot is preferred over re-publishing `start`, + # which every stream consumer would read as a second activation. + from acb_common import activity + + r = _FakePresenceRedis() + monkeypatch.setattr(activity, "_get_client", lambda: r) + + async def _run(): + await activity._axadd({ + "kind": "agent", "phase": "start", "run_id": "run-inst", + "agent": "email-assistant", "user": "alice@fracktal.in", + "source": "email", "ts": "2026-08-02T00:00:00+00:00", + }) + # The pre-fix state, asserted so the regression is visible if the + # refresh is ever removed. + before = json.loads(r.store[activity._live_key("run-inst")]) + assert "instance" not in before + + activity.refresh_run_presence("run-inst", instance="u:alice@fracktal.in") + await asyncio.sleep(0) + await asyncio.sleep(0) + # …and the office view's own reader sees it. + return await activity.active_runs() + + live = asyncio.run(_run()) + snap = json.loads(r.store[activity._live_key("run-inst")]) + assert snap["instance"] == "u:alice@fracktal.in" + # Nothing else in the snapshot was disturbed — it is a patch, not a rewrite. + assert snap["agent"] == "email-assistant" + assert snap["user"] == "alice@fracktal.in" + assert snap["phase"] == "start" + assert len(r.xadded) == 1, "the refresh must not add a stream entry" + assert [e["instance"] for e in live] == ["u:alice@fracktal.in"] + + +def test_presence_refresh_never_resurrects_a_finished_run(monkeypatch): + # A patch that lands after the end event (or for a run that never started) + # must be a no-op — a resurrected presence key would show a finished agent + # as working for LIVE_TTL_SECONDS. + from acb_common import activity + + r = _FakePresenceRedis() + monkeypatch.setattr(activity, "_get_client", lambda: r) + + async def _run(): + await activity._axadd({ + "kind": "agent", "phase": "start", "run_id": "run-gone", + }) + await activity._axadd({ + "kind": "agent", "phase": "end", "run_id": "run-gone", + }) + activity.refresh_run_presence("run-gone", instance="u:a@b.com") + await asyncio.sleep(0) + await asyncio.sleep(0) + + asyncio.run(_run()) + assert r.store == {} + + +def test_presence_refresh_ignores_empty_and_missing_input(monkeypatch): + # A shared run's instance is '' — nothing to patch, and no Redis touched + # (not even the no-running-loop one-shot path, which would open a client). + from acb_common import activity + + calls: list[tuple] = [] + + async def _spy(r, run_id, patch): + calls.append((run_id, patch)) + + monkeypatch.setattr(activity, "_apply_presence_patch", _spy) + activity.refresh_run_presence("run-x", instance="") + activity.refresh_run_presence("run-x", instance=None) + activity.refresh_run_presence("", instance="u:a@b.com") + assert calls == [] + + def test_model_events_never_create_a_presence_key(monkeypatch): # kind="model" is cost/feed-only — it must never leak into presence, even # though the gate now accepts more than one kind. @@ -245,6 +367,8 @@ async def hgetall(self, key): "agent|orchestrator|cost": "0.35", "agent|orchestrator|calls": "2", "agent|email-assistant|cost": "0.15", + "instance|u:alice@fracktal.in|cost": "0.15", + "instance|u:alice@fracktal.in|calls": "1", } return {} @@ -264,6 +388,67 @@ async def hgetall(self, key): assert out["by_agent"]["orchestrator"]["cost"] == 0.35 assert out["by_agent"]["orchestrator"]["calls"] == 2 assert out["by_agent"]["email-assistant"]["cost"] == 0.15 + # WS-6a interim: "what did alice's personal agents cost today" is a rollup + # dimension, not a scan of the (2000-entry-bounded) raw stream. + assert out["by_instance"]["u:alice@fracktal.in"]["cost"] == 0.15 + assert out["by_instance"]["u:alice@fracktal.in"]["calls"] == 1 + + +def test_cost_rollup_folds_the_tenant_partition(monkeypatch): + # The write side of the same dimension: a priced model event carrying an + # instance increments instance||cost, and a shared run (no instance) + # writes no such field at all — byte-identical to the pre-WS-6 rollup. + from acb_common import activity + + class _FakePipe: + def __init__(self): + self.ops: list[tuple] = [] + + def hincrbyfloat(self, key, field, val): + self.ops.append((field, val)) + + def hincrby(self, key, field, val): + self.ops.append((field, val)) + + def expire(self, key, ttl): + pass + + async def execute(self): + return [] + + class _FakeRedis: + def __init__(self): + self.pipe = _FakePipe() + + def pipeline(self, transaction=False): + return self.pipe + + r = _FakeRedis() + asyncio.run(activity._record_cost(r, { + "kind": "model", "model": "gpt-4o-mini", "cost_usd": 0.25, + "tokens": 100, "source": "email", "agent": "email-assistant", + "instance": "u:alice@fracktal.in", + })) + fields = dict(r.pipe.ops) + assert fields["instance|u:alice@fracktal.in|cost"] == 0.25 + assert fields["instance|u:alice@fracktal.in|calls"] == 1 + + shared = _FakeRedis() + asyncio.run(activity._record_cost(shared, { + "kind": "model", "model": "gpt-4o-mini", "cost_usd": 0.25, + "tokens": 100, "source": "chat", "agent": "task-manager", + })) + assert not [f for f, _ in shared.pipe.ops if f.startswith("instance|")] + + +def test_split_field_parses_an_instance_key(): + # Partition keys contain ':' and '@' but never '|', so the dim|name|metric + # parse is unambiguous. + from acb_common.activity import _split_field + assert _split_field("instance|u:alice@fracktal.in|cost") == ( + "instance", "u:alice@fracktal.in", "cost") + assert _split_field("instance|t:growth|calls") == ( + "instance", "t:growth", "calls") def test_cost_summary_empty_history_is_all_zero(monkeypatch): diff --git a/tests/unit/test_instance_wiring.py b/tests/unit/test_instance_wiring.py index 71a2130b9..783b4f8da 100644 --- a/tests/unit/test_instance_wiring.py +++ b/tests/unit/test_instance_wiring.py @@ -65,6 +65,76 @@ def test_unparseable_config_means_shared() -> None: ) in ("", ) +def test_the_run_context_carries_the_key_the_executor_resolved() -> None: + """WS-6a: the same key that partitions disk + store also stamps telemetry. + + ``_bind_run_instance`` is the executor's top-up bind, issued right after + ``_resolve_agent_instance`` (which needs ``loaded.config``, so it cannot + run at the run boundary where the other correlation fields are bound). + """ + from acb_common import clear_run_context, get_run_context + from orchestrator.executor import ( + _bind_run_instance, + _resolve_agent_instance, + ) + + key = _resolve_agent_instance( + PERSONAL_CFG, "email-assistant", "alice@fracktal.in", + ) + _bind_run_instance(key) + try: + assert key == "u:alice@fracktal.in" + assert get_run_context()["instance"] == key + finally: + clear_run_context() + assert "instance" not in get_run_context() + + +def test_the_late_bind_also_patches_the_live_presence_key(monkeypatch) -> None: + """WS-6a: `phase="start"` is published before the load, so the presence key + it writes has no instance. `_bind_run_instance` patches that snapshot — + otherwise /observability/active and the office roster could never show a + partition for ANY run. A shared run patches nothing.""" + import acb_common + from acb_common import clear_run_context + from orchestrator.executor import _bind_run_instance + + patched: list[tuple] = [] + # The executor imports the name inside the call, so patching the package + # attribute is what intercepts it. + monkeypatch.setattr( + acb_common, "refresh_run_presence", + lambda run_id, **f: patched.append((run_id, f)), + ) + + try: + _bind_run_instance("u:alice@fracktal.in", "run-77") + assert patched == [("run-77", {"instance": "u:alice@fracktal.in"})] + _bind_run_instance("", "run-78") # shared → nothing to patch + _bind_run_instance("u:alice@fracktal.in") # no run_id → no presence key + assert len(patched) == 1 + finally: + clear_run_context() + + +def test_a_shared_run_stamps_no_partition() -> None: + """The shared key is '' — bound as an ABSENT field, not an empty one, so a + shared agent's telemetry is byte-identical to its pre-WS-6 shape.""" + from acb_common import clear_run_context, get_run_context + from orchestrator.executor import ( + _bind_run_instance, + _resolve_agent_instance, + ) + + _bind_run_instance( + _resolve_agent_instance(SHARED_CFG, "task-manager", "alice@fracktal.in") + ) + try: + assert "instance" not in get_run_context() + finally: + clear_run_context() + + def test_directory_precedence_is_exact(tmp_path: Path, monkeypatch) -> None: """session override > instance > workspace_root > clone dir.""" from orchestrator.executor import _resolve_effective_agent_dir diff --git a/tests/unit/test_llm_usage_telemetry.py b/tests/unit/test_llm_usage_telemetry.py index 268a5f742..d34f51b65 100644 --- a/tests/unit/test_llm_usage_telemetry.py +++ b/tests/unit/test_llm_usage_telemetry.py @@ -132,6 +132,133 @@ def test_emit_usage_never_raises_on_bad_response(): llm_client._emit_usage("gpt-4o-mini", "", {"usage": None}, source="memory") +# ── WS-6c: the D1 attribution four-tuple (run_id, member, agent, instance) ── +# These assert on the event as it reaches the STREAM (_axadd), not on the +# publish_activity kwargs — the whole point is that the run context, not the +# call site, supplies the stamp, and that merge happens inside _build_event. + +_USAGE = {"usage": {"prompt_tokens": 1000, "completion_tokens": 500, + "total_tokens": 1500}} + + +def _emit_and_capture(monkeypatch, *, bind: dict | None = None, **emit_kw): + """Run _emit_usage on a loop with _axadd stubbed; return the events.""" + import asyncio + + from acb_common import activity, bind_run_context, clear_run_context + + recorded: list[dict] = [] + + async def _fake_axadd(evt): + recorded.append(evt) + + monkeypatch.setattr(activity, "_axadd", _fake_axadd) + + async def _run(): + if bind: + bind_run_context(**bind) + try: + llm_client._emit_usage("gpt-4o-mini", "tier-fast", _USAGE, **emit_kw) + await asyncio.sleep(0) + await asyncio.sleep(0) + finally: + clear_run_context() + + asyncio.run(_run()) + return recorded + + +def test_in_run_call_is_attributed_with_no_call_site_change(monkeypatch): + # THE acceptance criterion: _emit_usage is called exactly as it always was + # — no attribution kwargs at all — and the activation still carries all + # four D1 fields, because they ride the run context. + events = _emit_and_capture(monkeypatch, bind={ + "run_id": "run-42", "thread_id": "t-1", "agent": "email-assistant", + "user": "alice@fracktal.in", "source": "email", + "instance": "u:alice@fracktal.in", + }) + assert len(events) == 1 + ev = events[0] + assert ev["run_id"] == "run-42" + assert ev["user"] == "alice@fracktal.in" # the member + assert ev["agent"] == "email-assistant" + assert ev["instance"] == "u:alice@fracktal.in" # the tenant partition + # …without losing anything the feed already showed. + assert ev["kind"] == "model" + assert ev["model"] == "gpt-4o-mini" + assert ev["source"] == "email" + assert ev["tokens"] == 1500 + + +def test_shared_agent_run_emits_no_instance_field(monkeypatch): + # '' (shared) is absent, never the string "''" — the event shape a + # pre-WS-6 consumer already handles. + events = _emit_and_capture(monkeypatch, bind={ + "run_id": "run-43", "agent": "task-manager", "source": "chat", + "instance": "", + }) + assert "instance" not in events[0] + + +def test_out_of_run_caller_may_pass_the_stamp_explicitly(monkeypatch): + # The v1_compat-shaped case (WS-6b's future call site): a bare HTTP request + # inherits no contextvars, so it supplies the tuple itself. `member` lands + # on the feed's `user` field so both paths roll up identically. + events = _emit_and_capture( + monkeypatch, + source="chat", agent="orchestrator", + run_id="run-44", member="bob@fracktal.in", instance="t:growth", + ) + ev = events[0] + assert ev["run_id"] == "run-44" + assert ev["user"] == "bob@fracktal.in" + assert ev["agent"] == "orchestrator" + assert ev["instance"] == "t:growth" + + +def test_existing_source_and_agent_kwargs_are_untouched(monkeypatch): + # No-regression: today's only explicit caller (v1_compat) passes exactly + # these two and must keep behaving identically. + events = _emit_and_capture(monkeypatch, source="chat", agent="apis-config") + ev = events[0] + assert ev["source"] == "chat" + assert ev["agent"] == "apis-config" + assert "run_id" not in ev and "instance" not in ev + + +def test_unpriced_model_publishes_unknown_cost_never_zero(monkeypatch): + # The null-cost contract, end to end through the emitter: _compute_cost + # returns None for a stub-priced model and NOTHING may coerce that to 0 — + # a confident $0.00 is worse than an honest "—", and a 0 would also fold + # into the daily rollup as real spend. + from acb_llm.client import ensure_model_registered + + model = "deepseek/deepseek-v98-unpriced-emit-test" + ensure_model_registered(model) + + import asyncio + + from acb_common import activity + + recorded: list[dict] = [] + + async def _fake_axadd(evt): + recorded.append(evt) + + monkeypatch.setattr(activity, "_axadd", _fake_axadd) + + async def _run(): + llm_client._emit_usage(model, "tier-fast", _USAGE) + await asyncio.sleep(0) + await asyncio.sleep(0) + + asyncio.run(_run()) + ev = recorded[0] + assert ev.get("cost_usd") is None + assert ev.get("cost_usd") != 0 + assert "cost_usd" not in ev # dropped, so the rollup skips it entirely + + # ── App-source inference (zero-touch cross-app attribution) ────────────────── def test_infer_app_source_reads_the_originating_app_module(): diff --git a/tests/unit/test_observability.py b/tests/unit/test_observability.py index c5cbce3f9..28c5677b8 100644 --- a/tests/unit/test_observability.py +++ b/tests/unit/test_observability.py @@ -51,11 +51,87 @@ def test_bound_context_visible_to_any_emitter_on_the_task(): def test_clear_removes_context_no_leak(): - bind_run_context(run_id="r1", agent="a") + bind_run_context(run_id="r1", agent="a", instance="u:a@b.com") clear_run_context() merged = structlog.contextvars.get_contextvars() assert "run_id" not in merged assert "agent" not in merged + # WS-6a: the tenant partition must not survive into the next run either — + # a leaked instance would attribute one member's spend to another's. + assert "instance" not in merged + + +# ── WS-6a: the tenant partition is part of the run stamp ──────────────────── +# `instance` completes decision D1's (run_id, member, agent, instance) tuple. +# Its vocabulary is the manifest's (AgentManifest.instance_key): '' shared, +# u: personal, t: team — the same key agent_blob carries. + +def test_instance_is_bound_and_readable_like_every_other_run_field(): + bind_run_context( + run_id="r-i", agent="email-assistant", user="alice@fracktal.in", + instance="u:alice@fracktal.in", + ) + try: + assert get_run_context()["instance"] == "u:alice@fracktal.in" + # And it reaches the log line the same way the others do. + assert structlog.contextvars.get_contextvars()["instance"] == ( + "u:alice@fracktal.in" + ) + finally: + clear_run_context() + + +def test_shared_agent_binds_no_instance_key_at_all(): + # '' is how the manifest/store spell "shared". It must land as an ABSENT + # key — never the empty string, and never the literal "''" (a quoted empty + # value in the feed would read as a real, unreachable partition). + bind_run_context(run_id="r-s", agent="task-manager", instance="") + try: + ctx = get_run_context() + assert ctx == {"run_id": "r-s", "agent": "task-manager"} + assert "instance" not in ctx + finally: + clear_run_context() + + +def test_inherit_and_run_context_keys_match(): + """Drift gate: the twin tuples must be extended together. + + ``_log._RUN_CONTEXT_KEYS`` decides what a run BINDS; ``activity._INHERIT`` + decides what an event whose emitter omitted a field COPIES from that bind. + A key in one and not the other half-lands: bound onto every log line but + missing from every activity/cost event, or the reverse. Both AGENTS.md + files that document the stamp say "extend those two tuples together" — + this is the assertion behind that sentence. + """ + from acb_common import _log, activity + + bound = set(_log._RUN_CONTEXT_KEYS) + inherited = set(activity._INHERIT) + assert inherited == bound, ( + "acb_common.activity._INHERIT and acb_common._log._RUN_CONTEXT_KEYS " + f"have drifted: bound-but-not-inherited={sorted(bound - inherited)}, " + f"inherited-but-not-bound={sorted(inherited - bound)}. Add the key to " + "BOTH tuples (and to bind_run_context's signature + its field list), " + "or attribution half-lands — see packages/AGENTS.md and " + "specs/observability_e2.md §7." + ) + + +def test_a_second_bind_tops_up_instance_without_disturbing_the_first(): + # The ordering fix WS-6a ships: the run boundary binds early (so a failure + # during agent load is still correlated), then _bind_run_instance tops up + # `instance` once loaded.config makes it resolvable. + bind_run_context(run_id="r-2b", thread_id="t", agent="a", user="u@x", + source="chat") + try: + bind_run_context(instance="t:growth") + assert get_run_context() == { + "run_id": "r-2b", "thread_id": "t", "agent": "a", "user": "u@x", + "source": "chat", "instance": "t:growth", + } + finally: + clear_run_context() def test_get_run_context_reflects_bound_fields():