diff --git a/ROADMAP.md b/ROADMAP.md index 90cc39a..ada7bcf 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -454,7 +454,7 @@ allowances; the agent records document-driven denials and approval decisions. - [x] **10.4 — Cost attribution** per run, thread (session), node and explicit tenant. The price is now recorded, not guessed: every gateway publishes a `cost_usd` through the same `llm_output` envelope, the runtime's usage - callback accumulates it per node and writes it onto the `end` event, and + callback accumulates it per node and writes it onto the `end` or `error` event, and an `AgentNode` writes the per-call figure onto each `model` event. A backend that reports no price still falls back to a `RateCard` estimate, and `recorded_cost_usd` and `estimated_cost_usd` never mix — a recorded @@ -467,9 +467,10 @@ allowances; the agent records document-driven denials and approval decisions. `grapharc cost --tenant NAME` select whole labelled runs, refuse conflicting labels, and leave old unlabelled runs unassigned. Planning tokens without a recorded price are estimated or marked unpriced, including in tenant - totals. **Producer limit:** planner turns and failed node terminal events - still omit the provider price ([#138](https://github.com/CodeGraphContext/GraphARC/issues/138)); - a rate-card estimate is not a recorded bill. + totals. Planner turns retain the call-specific price and model from the + returned message envelope, including parse failures. Every node terminal + error path retains its own usage scope, excluding concurrent siblings' + spend. A rate-card estimate is not a recorded bill. Tenant labels provide attribution, not authorization. - [ ] **10.2 — Rollback** and versioned graph/prompt configs. - [ ] **10.5 — Alerting** on budget, failure, and verifier-drift. diff --git a/docs/cookbook/07-slack.md b/docs/cookbook/07-slack.md index 0703b6e..1f5c3c8 100644 --- a/docs/cookbook/07-slack.md +++ b/docs/cookbook/07-slack.md @@ -166,11 +166,22 @@ status message is posted to the channel (a `respond()`-style reply would allow only five updates), so a live run is visible to everyone in it — the mention path threads it under your message as before. -Because the trace now lands inside the working directory, the run is also -inspectable afterwards from Slack itself: `/grapharc metrics -slack-runs//trace.jsonl `, `viz` for the finished diagram, -`replay` for the reconstruction. The `slack-runs/` directories are the audit -trail and are never cleaned up automatically; prune them like any other logs. +Because the trace lands inside the working directory, the run is also +inspectable afterwards from Slack itself: `/grapharc metrics`, `viz` for the +finished diagram, `replay` for the reconstruction, or `trace` for its events. +Bare readers use the latest completed trace remembered for that channel. In a +new channel or after a bot restart, they fall back to the newest trace under +`slack-runs/`. Each reply names the selected path, run id and whether it used +that workdir fallback. A trace containing zero or multiple runs needs explicit +selection; the bot does not guess. With no previous trace, it suggests +`demo stage0` instead of returning an argparse usage error. + +Explicit commands such as `/grapharc metrics +slack-runs//trace.jsonl ` keep working. `diff` still needs an +explicit path and two run ids. Inferred paths pass the same confinement gate +as typed paths. Channel history is process-local; the traces are durable. The +`slack-runs/` directories are the audit trail and are never cleaned up +automatically; prune them like any other logs. ## Watching it live in a browser diff --git a/docs/deep-dive.md b/docs/deep-dive.md index cfba7da..1f86ff9 100644 --- a/docs/deep-dive.md +++ b/docs/deep-dive.md @@ -94,7 +94,7 @@ OpenRouter also carries routing: model-level `fallback_models` chains, provider **Cost ceilings are enforced, on both sides of a call.** A `SpendMeter` refuses before a call once the ceiling is reached, and after a call it charges the cost and *then* raises if that crossed the line — so overspend is bounded by the single call that crossed it rather than discovered a node later. The limit worth stating: a call whose cost the provider does not report cannot be charged, and those land in `unpriced_calls` rather than being guessed at. `unpriced_calls > 0` means the ceiling saw less than the whole bill. -Caveats each backend accepts openly. **Claude CLI:** `bind_tools` and `with_structured_output` raise `NotImplementedError` — the adapter implements neither, so what you get is LangChain's `BaseChatModel` default, and `claude -p` in the tool-free mode GraphARC drives it in offers nothing to implement them with. GraphARC also has no cache control on this path — the CLI decides and the usage envelope reports what it did — and calls spend subscription quota. **OpenRouter:** credit is reserved against `max_tokens`, so the default is deliberately modest. **OpenAI:** the API returns token counts and no price, so `cost_usd` is `None` and a dollar ceiling counts calls instead of enforcing — pass `price_per_million=` or price the trace afterwards with `observe.cost.RateCard`; token budgets are unaffected. **Ollama:** free by definition, so calls are charged `0.0` rather than counted as unpriced, and whether tool-calling works depends on the model you pulled rather than on the adapter. **All of them:** a reported per-call `cost_usd` is captured and budgeted against. Successful node events and agent model events carry it into the trace; planner turns and failed node terminal events currently omit it. See [ROADMAP.md](../ROADMAP.md) §10.4 for explicit tenant attribution and producer pricing limits. +Caveats each backend accepts openly. **Claude CLI:** `bind_tools` and `with_structured_output` raise `NotImplementedError` — the adapter implements neither, so what you get is LangChain's `BaseChatModel` default, and `claude -p` in the tool-free mode GraphARC drives it in offers nothing to implement them with. GraphARC also has no cache control on this path — the CLI decides and the usage envelope reports what it did — and calls spend subscription quota. **OpenRouter:** credit is reserved against `max_tokens`, so the default is deliberately modest. **OpenAI:** the API returns token counts and no price, so `cost_usd` is `None` and a dollar ceiling counts calls instead of enforcing — pass `price_per_million=` or price the trace afterwards with `observe.cost.RateCard`; token budgets are unaffected. **Ollama:** free by definition, so calls are charged `0.0` rather than counted as unpriced, and whether tool-calling works depends on the model you pulled rather than on the adapter. **All of them:** a reported per-call `cost_usd` is captured and budgeted against. Successful and failed node terminal events, planner turns and agent model events carry it into the trace. See [ROADMAP.md](../ROADMAP.md) §10.4 for explicit tenant attribution and producer pricing limits. ## Tools and the harness @@ -248,7 +248,7 @@ A stable system is not one that claims to have no edges — it is one whose edge - **`run_command` is not confined.** Argv-only and never a shell, but the child is an ordinary process with your privileges. - **`interrupt()` suspends but cannot be resumed.** LangGraph's native interrupt stops the graph and shows on `get_state`, and there is no supported resume path — resuming means passing a `Command` as *input*, which is closed by design. Use the session layer's approval gate for human-in-the-loop. - **Still unwrapped from LangGraph:** `retry_policy`, `cache_policy`, `durability`, subgraphs. `.inner` reaches them, but execution entry points there fail closed, so `.inner` is an inspection escape hatch and not a way to run the graph. -- **Cost is recorded when a backend reports one, estimated when it does not.** Both gateways publish the provider's `cost_usd` through the same `llm_output` envelope, the runtime's usage callback writes it onto the node's `end` event, and an agent's `model` events carry the per-call breakdown. A backend that reports no price still falls back to a `RateCard` estimate, and the two figures stay apart — `recorded_cost_usd` is never a guess. Explicit tenants are supported by `attribute_tenant` and `grapharc cost`; legacy unlabelled runs are excluded from named tenant bills. Planner turns and failed node terminal events still omit the provider price, so their tokens need a rate card or remain unpriced. +- **Cost is recorded when a backend reports one, estimated when it does not.** Both gateways publish the provider's `cost_usd` through the same `llm_output` envelope, the runtime's usage callback writes it onto the node's `end` event, and an agent's `model` events carry the per-call breakdown. A backend that reports no price still falls back to a `RateCard` estimate, and the two figures stay apart — `recorded_cost_usd` is never a guess. Explicit tenants are supported by `attribute_tenant` and `grapharc cost`; legacy unlabelled runs are excluded from named tenant bills. Planner turns and all failed node terminal paths retain the provider price and model identity when reported; parallel node failures retain only their own token spend. - **A node's tokens are its own, not the run's movement while it ran.** Worth stating because it was the other way round: an `end` event carried the difference between two readings of the run's *shared* meter, so under fan-out the workers' windows overlapped and each was credited with its siblings' concurrent spend. Three workers costing 8 tokens each traced as 24/16/8, and `metrics` and `cost` agreed on 48 for 24 tokens of real work — doubling the estimated bill purely because the work ran in parallel. Attribution now comes from a per-node scope on the meter, so the same work costs the same serially and in parallel; a hand charge the usage callback never saw still lands on the node that made it. - **A planning round is an envelope, not a measurement.** A `round` event used to carry the planner's `tokens` and the round's `duration_ms`, both of which `metrics`, `cost` and `replay` add on top of node totals — and the planner's spend was already reported by its own `plan` event, so it was counted twice, and a round's duration encloses the plan plus every node it ran. Neither is on the event now; both are on its `state_delta` as `round_tokens` / `round_iterations` / `round_duration_ms`, where no reader sums them. `RoundRecord.iterations` also holds a figure now rather than always `0`. - **The Claude CLI backend is completion-only, and an agent node on it is *delegated* rather than governed.** The CLI has no tool-calling wire format, so GraphARC cannot run its own gated loop over it. Rather than refuse, `AgentNode` hands the whole loop to Claude Code's headless agent, in one of two named tiers. `allowlist`, the default, pre-approves exactly the Claude Code twins of the node's own registered tools (`read_file`→`Read`, … `run_command`→`Bash`), so one operator declaration governs both the governed loop and the delegated one — but the enforcement is Claude Code's own gating, not this graph's per-call policy, there are no per-tool trace events, and anything unlisted falls to headless default gating, which fails closed. `bypass`, explicit opt-in only, runs `bypassPermissions`: every tool Claude Code has, no checks at all. In either tier the calls are not confined by the sandbox executor and the token figure is the sub-agent's own rather than one GraphARC metered call by call; a `--max-tokens` the delegated path cannot enforce is refused rather than silently unapplied. The workspace boundary and the wall-clock ceiling still hold — the CLI runs in its own session, and the deadline kills the whole process group, not just the direct child. It warns on `DelegatedToolUseWarning` at construction, naming the tier, and marks every trace event `executor=delegated` with its `delegated_mode`; filter that warning to an error to get the old refusal back. Structured output still needs an OpenAI-wire backend: `openrouter`, `openai`, or a local `ollama`. @@ -256,7 +256,7 @@ A stable system is not one that claims to have no edges — it is one whose edge - **`.env` and `grapharc.toml` follow the same discovery rule: the working directory, and nowhere else.** Neither searches parent directories — a run must not be governed by a file you did not know about, and must not be *billed* to one either. **This is a behaviour change:** the credential loader used to walk up to `/`, so a `.env` in an ancestor directory (a `$HOME` one on a shared box, a client project one above a demo checkout) was picked up silently. If you relied on that, move the file into the directory you run from, `export` the variable, or pass `env_file=` to name it explicitly. A real environment variable still beats any file. - **`grapharc run` has no budget unless you give it one.** Set any of `--max-tokens`, `--max-iterations`, `--max-seconds`, or `--max-concurrency`; without them each dimension is unlimited and the gate admits a topology of any worst-case cost. -**Verified this pass:** `pytest` → green, 2,354 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. +**Verified this pass:** `pytest` → green, 2,401 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. [ROADMAP.md](../ROADMAP.md) tracks what is built and what is not, item by item. diff --git a/grapharc/planner/proposal.py b/grapharc/planner/proposal.py index 40954c4..7c953b8 100644 --- a/grapharc/planner/proposal.py +++ b/grapharc/planner/proposal.py @@ -690,10 +690,19 @@ def propose( error = "" proposal = self._stamp(proposal) + # Read this call's envelope, rather than a model's mutable last_usage: + # the latter can describe another concurrent call, or an earlier call + # when this one failed before producing a message. + metadata = raw_message.response_metadata if raw_message is not None else {} + cost = (metadata.get("token_usage") or {}).get("cost") + if cost is None: + cost = metadata.get("cost_usd") self._emit( ctx, duration_ms=(time.perf_counter() - started) * 1000, tokens=tokens, + cost_usd=cost, + model=metadata.get("model_name"), state_delta={ "structured": structured, "nodes": proposal.node_count() if proposal else 0, diff --git a/grapharc/runtime/graph.py b/grapharc/runtime/graph.py index 02b7c7a..e02155e 100644 --- a/grapharc/runtime/graph.py +++ b/grapharc/runtime/graph.py @@ -855,6 +855,7 @@ def _leave( usage: Any = None, ) -> Any: """Everything both node wrappers do after the node body returns.""" + measured = _usage_fields(ctx, tokens_before, usage) delta: dict[str, Any] | None = None if result is not None: try: @@ -864,7 +865,7 @@ def _leave( "error", duration_ms=duration_ms, error=str(err), - tokens=ctx.meter.tokens - tokens_before, + **measured, ) raise @@ -881,31 +882,15 @@ def _leave( "error", duration_ms=duration_ms, error=f"budget: {exc.reason}", - tokens=ctx.meter.tokens - tokens_before, + **measured, ) raise - # The provider's own price for every model call made inside this node, - # collected by the usage callback. Left None when no backend reported - # one, so `observe.cost` can keep a measured figure apart from a - # rate-card estimate instead of recording a guess as a fact. - models = getattr(usage, "models", ()) - # This node's own spend, from the meter's per-node scope, not the movement - # of the run's shared total while the node ran. Under fan-out those differ: - # the workers' windows overlap, so every worker used to be credited with - # each sibling's concurrent spend — three workers costing 8 tokens each - # traced as 24/16/8, and `metrics` and `cost` both reported 48 for 24 - # tokens of real work, doubling the estimated bill purely because the work - # ran in parallel. Falls back to the difference when no scope was open, so - # a caller driving `_leave` outside `charging()` still gets a figure. - spent = getattr(usage, "tokens", None) emit( "end", state_delta=delta, duration_ms=duration_ms, - tokens=ctx.meter.tokens - tokens_before if spent is None else spent, - cost_usd=getattr(usage, "cost_usd", None), - model=models[0] if len(models) == 1 else None, + **measured, ) return result @@ -935,7 +920,7 @@ async def awrapped(state: Any, config: RunnableConfig) -> Any: "error", duration_ms=(time.perf_counter() - t0) * 1000, error=repr(exc), - tokens=ctx.meter.tokens - tokens_before, + **_usage_fields(ctx, tokens_before, usage), ) raise return self._leave( @@ -974,7 +959,7 @@ def wrapped(state: Any, config: RunnableConfig) -> Any: "error", duration_ms=(time.perf_counter() - t0) * 1000, error=repr(exc), - tokens=ctx.meter.tokens - tokens_before, + **_usage_fields(ctx, tokens_before, usage), ) raise return self._leave( @@ -991,6 +976,22 @@ def wrapped(state: Any, config: RunnableConfig) -> Any: return wrapped +def _usage_fields(ctx: RunContext, tokens_before: int, usage: Any) -> dict[str, Any]: + """The same measured spend on every terminal event, successful or failed. + + A node's own scope excludes concurrent siblings' spend. The shared meter + difference is only a fallback for callers outside `charging()`. A missing + provider price stays missing; rate-card estimates belong to trace readers. + """ + spent = getattr(usage, "tokens", None) + models = getattr(usage, "models", ()) + return { + "tokens": ctx.meter.tokens - tokens_before if spent is None else spent, + "cost_usd": getattr(usage, "cost_usd", None), + "model": models[0] if len(models) == 1 else None, + } + + def topology_delta(arc: GraphARC) -> dict[str, Any]: """The declared shape of a graph, as one trace-event `state_delta`. diff --git a/grapharc/slack/bot.py b/grapharc/slack/bot.py index c658c01..0d6729c 100644 --- a/grapharc/slack/bot.py +++ b/grapharc/slack/bot.py @@ -44,6 +44,7 @@ from grapharc.slack.config import SlackBotConfig from grapharc.slack.format import format_result, live_view_url, mermaid_live_url from grapharc.slack.live import ApprovalPrompt, LiveSettings, LiveSink, LiveTail +from grapharc.slack.recent import RecentRuns from grapharc.slack.runner import run_command #: `action_id`s for the two approval buttons. Stable strings, because a bolt @@ -61,7 +62,14 @@ def handle_text(text: str, config: SlackBotConfig) -> str: return handle_text_live(text, config, sink=None) -def handle_text_live(text: str, config: SlackBotConfig, sink: LiveSink | None) -> str: +def handle_text_live( + text: str, + config: SlackBotConfig, + sink: LiveSink | None, + *, + history: RecentRuns | None = None, + channel: str = "", +) -> str: """Like `handle_text`, but narrates a tracing command through `sink`. Returns the message the caller must still post — `""` when the sink @@ -70,6 +78,7 @@ def handle_text_live(text: str, config: SlackBotConfig, sink: LiveSink | None) - result always reaches the requester through one route or the other. """ stripped = _MENTION.sub("", text).strip() + history = history or RecentRuns() try: argv = parse_command( stripped, @@ -79,6 +88,11 @@ def handle_text_live(text: str, config: SlackBotConfig, sink: LiveSink | None) - timeout_seconds=config.timeout_seconds, work_timeout_seconds=config.work_timeout_seconds, ) + argv, resolved_note = history.resolve(argv, config.workdir, channel) + if resolved_note: + # Inferred arguments pass the exact same allowlist and path gate + # as typed arguments, including run ids that resemble flags. + argv = parse_command(shlex.join(argv), workdir=config.workdir) except SlackCommandError as exc: return str(exc) @@ -109,7 +123,10 @@ def handle_text_live(text: str, config: SlackBotConfig, sink: LiveSink | None) - if handle is None: result = run_command(argv, workdir=config.workdir, timeout_seconds=timeout) - return _with_final_links(format_result(result), argv, tpath, config, prior_runs) + history.remember(config.workdir, channel, tpath) + return resolved_note + _with_final_links( + format_result(result), argv, tpath, config, prior_runs + ) def _update(message: str, prompt: ApprovalPrompt | None = None) -> bool: try: @@ -129,6 +146,7 @@ def _update(message: str, prompt: ApprovalPrompt | None = None) -> bool: mutating_kinds=mutating_kinds_for(argv, config.workdir), ): result = run_command(argv, workdir=config.workdir, timeout_seconds=timeout) + history.remember(config.workdir, channel, tpath) final = _with_final_links(format_result(result), argv, tpath, config, prior_runs) if _update(final): return "" @@ -309,6 +327,7 @@ def build_app(config: SlackBotConfig) -> Any: ) from None app = App(token=config.bot_token) + history = RecentRuns() @app.command(config.slash_command) def _slash(ack: Any, respond: Any, command: dict[str, Any], client: Any) -> None: @@ -318,7 +337,9 @@ def _slash(ack: Any, respond: Any, command: dict[str, Any], client: Any) -> None return ack(f"running `grapharc {text}`…") sink = _ChannelSink(client, command.get("channel_id", "")) - reply = handle_text_live(text, config, sink) + reply = handle_text_live( + text, config, sink, history=history, channel=command.get("channel_id", "") + ) if reply: respond(reply) @@ -326,7 +347,10 @@ def _slash(ack: Any, respond: Any, command: dict[str, Any], client: Any) -> None def _mention(event: dict[str, Any], say: Any, client: Any) -> None: thread_ts = event.get("thread_ts") or event.get("ts") sink = _ChannelSink(client, event.get("channel", ""), thread_ts=thread_ts) - reply = handle_text_live(event.get("text", ""), config, sink) + reply = handle_text_live( + event.get("text", ""), config, sink, + history=history, channel=event.get("channel", ""), + ) if reply: say(reply, thread_ts=thread_ts) diff --git a/grapharc/slack/recent.py b/grapharc/slack/recent.py new file mode 100644 index 0000000..4330888 --- /dev/null +++ b/grapharc/slack/recent.py @@ -0,0 +1,96 @@ +"""Resolve bare Slack readers to an explicitly named recent run. + +Channel history is process-local. After restart, or in a new channel, readers +fall back to the newest trace under the bot's own `slack-runs/` directory and +say so. Explicit reader arguments never pass through this resolver. +""" + +from __future__ import annotations + +import threading +from pathlib import Path + +from grapharc.observe.trace import TraceReadError, TraceRecorder +from grapharc.slack.command import SlackCommandError, confine_path + +READERS = frozenset({"metrics", "viz", "replay", "trace"}) + + +class RecentRuns: + """Latest completed trace per channel, shared by a bot's worker threads.""" + + def __init__(self) -> None: + self._channels: dict[tuple[Path, str], Path] = {} + self._lock = threading.Lock() + + def remember(self, workdir: Path, channel: str, path: Path | None) -> None: + if not channel or path is None: + return + try: + workdir = workdir.resolve() + resolved = path.resolve() + confine_path(str(resolved), workdir) + if not resolved.is_relative_to((workdir / "slack-runs").resolve()): + return + if not resolved.is_file(): + return + except (SlackCommandError, OSError): + return + with self._lock: + self._channels[(workdir.resolve(), channel)] = resolved + + def resolve(self, argv: list[str], workdir: Path, channel: str = "") -> tuple[list[str], str]: + """Complete a bare reader or refuse an empty/ambiguous history.""" + if len(argv) != 1 or argv[0] not in READERS: + return argv, "" + workdir = workdir.resolve() + name = argv[0] + with self._lock: + selected = self._channels.get((workdir.resolve(), channel)) if channel else None + fallback = selected is None + if selected is not None: + confine_path(str(selected), workdir) + if not selected.is_file(): + selected = None + fallback = True + if selected is None: + root = workdir / "slack-runs" + confine_path(str(root), workdir) + candidates: list[tuple[int, str, Path]] = [] + try: + for path in root.glob("*/trace.jsonl"): + # Recheck before stat/read: generated paths and symlinks + # must obey the same boundary as explicitly typed paths. + try: + confine_path(str(path), workdir) + if path.is_file(): + candidates.append((path.stat().st_mtime_ns, str(path), path)) + except (SlackCommandError, OSError): + continue + except OSError as exc: + raise SlackCommandError(f"could not read recent Slack runs: {exc}") from None + if not candidates: + raise SlackCommandError( + f"No prior Slack run for `{name}`. Run `demo stage0` first, " + f"or supply `{name} path/to/trace.jsonl" + + (" run_id`." if name != "trace" else "`.") + ) + selected = max(candidates)[2] + confine_path(str(selected), workdir) + relative = str(selected.resolve().relative_to(workdir.resolve())) + try: + runs = TraceRecorder(selected).run_ids() + except (OSError, TraceReadError) as exc: + raise SlackCommandError( + f"cannot read the latest Slack trace `{relative}`: {exc}" + ) from None + if len(runs) != 1: + raise SlackCommandError( + f"Latest Slack trace `{relative}` contains {len(runs)} runs. " + f"Supply an explicit path and run id for `{name}`." + ) + run_id = runs[0] + resolved = [name, relative] + resolved.extend(["--run-id", run_id] if name == "trace" else [run_id]) + source = "workdir fallback" if fallback else "this channel" + return resolved, f"{name} for `{relative}` · run `{run_id}` ({source})\n" diff --git a/tests/test_provider_price_events.py b/tests/test_provider_price_events.py new file mode 100644 index 0000000..296e278 --- /dev/null +++ b/tests/test_provider_price_events.py @@ -0,0 +1,243 @@ +"""Provider-reported prices survive planning and failed governed work (#138).""" + +from __future__ import annotations + +import asyncio + +import pytest +from langchain_core.runnables import RunnableLambda +from langgraph.types import Command + +from grapharc.observe import RateCard, TraceRecorder, attribute, attribute_tenant +from grapharc.observe.cost import attribute_thread +from grapharc.observe.metrics import summarize +from grapharc.planner import PlannerNode +from grapharc.runtime.budget import Budget, BudgetExceeded, BudgetMeter +from grapharc.runtime.graph import ( + END, + START, + GraphARC, + GraphRoutingError, + RunContext, + StateTypeError, + WritePermissionError, +) +from grapharc.runtime.state import GraphARCState +from grapharc.testing import ScriptedChatModel + + +class State(GraphARCState): + answer: str = "" + proposal: object = None + count: int = 0 + + +def context(): + return RunContext( + run_id="run", thread_id="thread", tenant="acme", graph="g", meter=BudgetMeter(Budget()) + ) + + +def assert_price(trace, price): + run = attribute(trace, "run") + assert run.recorded_cost_usd == pytest.approx(price) + assert attribute_thread(trace, "thread").recorded_cost_usd == pytest.approx(price) + assert attribute_tenant(trace, "acme").recorded_cost_usd == pytest.approx(price) + assert run.unpriced_tokens == 0 + assert run.tokens == summarize(trace, "run").tokens > 0 + + +@pytest.mark.parametrize("reply", ['{"nodes": []}', "not a proposal"]) +@pytest.mark.parametrize("price", [None, 0.0, 0.003]) +@pytest.mark.parametrize("structured", [False, True]) +def test_planner_records_the_call_price_even_when_parsing_fails(tmp_path, reply, price, structured): + class StructuredModel(ScriptedChatModel): + def with_structured_output(self, schema, **kwargs): + return RunnableLambda( + lambda messages: { + "raw": self.invoke(messages), + "parsed": None, + "parsing_error": ValueError("parse failed"), + } + ) + + trace = TraceRecorder(tmp_path / "trace.jsonl") + model = StructuredModel(responses=[reply], cost_usd=price, model_name="acme/planner") + ctx = context() + result = PlannerNode(model, trace=trace, structured=structured).propose("plan", ctx) + assert result.ok == (reply.startswith("{")) + event, = trace.read_events() + assert event.phase == "plan" + assert event.cost_usd == price + assert ctx.meter.tokens == result.tokens == event.tokens > 0 + cost = attribute(trace, "run", rates=RateCard(default=2.0)) + if price is None: + assert cost.recorded_cost_usd is None + assert cost.estimated_cost_usd == pytest.approx(result.tokens / 1000 * 2) + else: + assert event.model == "acme/planner" + assert_price(trace, price) + assert cost.estimated_cost_usd is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_node", [False, True]) +@pytest.mark.parametrize("failure", ["body", "write", "type", "routing", "budget", "manual_budget"]) +async def test_all_failed_node_terminals_retain_provider_usage(tmp_path, async_node, failure): + trace = TraceRecorder(tmp_path / "trace.jsonl") + model = ScriptedChatModel(responses=["done"], cost_usd=0.002, model_name="acme/worker") + graph = GraphARC(State, name="g", trace=trace) + + def finish(ctx): + if failure == "body": + raise ValueError("failed after spending") + if failure == "write": + return {"undeclared": True} + if failure == "type": + return {"count": "not an integer"} + if failure == "routing": + return Command(goto="unknown") + if failure == "manual_budget": + ctx.meter.charge_tokens(100) + return {"answer": "done"} + + def body(state, ctx): + model.invoke("work") + return finish(ctx) + + async def abody(state, ctx): + await model.ainvoke("work") + return finish(ctx) + + graph.add_node("worker", abody if async_node else body, writes={"answer", "count"}) + graph.add_edge(START, "worker") + graph.add_edge("worker", END) + compiled = graph.compile() + budget = Budget(max_tokens=1 if failure == "budget" else 50) + expected = { + "body": ValueError, + "write": WritePermissionError, + "type": StateTypeError, + "routing": GraphRoutingError, + "budget": BudgetExceeded, + "manual_budget": BudgetExceeded, + }[failure] + with pytest.raises(expected): + await compiled.ainvoke( + {}, run_id="run", thread_id="thread", tenant="acme", budget=budget + ) + terminal, = [e for e in trace.read_events() if e.phase == "error"] + assert terminal.cost_usd == pytest.approx(0.002) + assert terminal.model == "acme/worker" + assert terminal.tokens == compiled.last_run.meter.tokens > 0 + assert_price(trace, 0.002) + + +def test_nested_planner_price_and_tokens_are_counted_once(tmp_path): + trace = TraceRecorder(tmp_path / "trace.jsonl") + model = ScriptedChatModel(responses=['{"nodes": []}'], cost_usd=0.003) + planner = PlannerNode(model, trace=trace, structured=False) + graph = GraphARC(State, name="g", trace=trace) + graph.add_node("planner", lambda state, ctx: {"proposal": planner.propose("plan", ctx)}, + writes={"proposal"}) + graph.add_edge(START, "planner") + graph.add_edge("planner", END) + compiled = graph.compile() + compiled.invoke({}, run_id="run", thread_id="thread", tenant="acme") + plan, = [e for e in trace.read_events() if e.phase == "plan"] + end, = [e for e in trace.read_events() if e.phase == "end"] + assert plan.cost_usd == end.cost_usd == pytest.approx(0.003) + assert plan.tokens == end.tokens == compiled.last_run.meter.tokens + assert_price(trace, 0.003) + + +def test_nested_planner_budget_failure_preserves_price_and_stops_more_calls(tmp_path): + trace = TraceRecorder(tmp_path / "trace.jsonl") + model = ScriptedChatModel(responses=['{"nodes": []}'] * 2, cost_usd=0.003) + planner = PlannerNode(model, trace=trace, structured=False) + graph = GraphARC(State, name="g", trace=trace) + + def body(state, ctx): + planner.propose("plan", ctx) + planner.propose("more", ctx) + return {} + + graph.add_node("planner", body, writes=set()) + graph.add_edge(START, "planner") + graph.add_edge("planner", END) + with pytest.raises(BudgetExceeded): + graph.compile().invoke({}, run_id="run", thread_id="thread", tenant="acme", + budget=Budget(max_tokens=1)) + assert model.call_count == 1 + assert_price(trace, 0.003) + + +@pytest.mark.asyncio +async def test_failed_parallel_node_is_not_billed_for_its_siblings(tmp_path): + trace = TraceRecorder(tmp_path / "trace.jsonl") + graph = GraphARC(State, name="g", trace=trace) + first = asyncio.Event() + second = asyncio.Event() + models = [ScriptedChatModel(responses=["done"], cost_usd=p) for p in (0.002, 0.005)] + + async def bad(state): + await models[0].ainvoke("first") + first.set() + await second.wait() + raise ValueError("failed after sibling spent") + + async def good(state): + await first.wait() + await models[1].ainvoke("second") + second.set() + return {} + + for name, body in [("bad", bad), ("good", good)]: + graph.add_node(name, body, writes=set()) + graph.add_edge(START, name) + graph.add_edge(name, END) + with pytest.raises(ValueError): + await graph.compile().ainvoke({}, run_id="run", thread_id="thread", tenant="acme") + terminal, = [e for e in trace.read_events() if e.phase == "error"] + assert terminal.tokens == models[0].last_usage["total_tokens"] + assert terminal.cost_usd == pytest.approx(0.002) + assert_price(trace, 0.007) + + +def test_a_failed_planner_call_cannot_reuse_an_earlier_calls_price(tmp_path): + trace = TraceRecorder(tmp_path / "trace.jsonl") + model = ScriptedChatModel(responses=['{"nodes": []}'], cost_usd=0.003) + planner = PlannerNode(model, trace=trace, structured=False) + ctx = context() + assert planner.propose("plan", ctx).ok + assert not planner.propose("no more replies", ctx).ok + first, failed = trace.read_events() + assert first.cost_usd == pytest.approx(0.003) + assert failed.cost_usd is None + assert failed.model is None + assert failed.tokens == 0 + assert_price(trace, 0.003) + + +def test_unpriced_failed_node_remains_unpriced_unless_a_rate_is_supplied(tmp_path): + trace = TraceRecorder(tmp_path / "trace.jsonl") + model = ScriptedChatModel(responses=["done"]) + graph = GraphARC(State, name="g", trace=trace) + + def body(state): + model.invoke("work") + raise ValueError("unpriced failure") + + graph.add_node("worker", body, writes=set()) + graph.add_edge(START, "worker") + graph.add_edge("worker", END) + with pytest.raises(ValueError): + graph.compile().invoke({}, run_id="run") + terminal, = [e for e in trace.read_events() if e.phase == "error"] + assert terminal.cost_usd is None + cost = attribute(trace, "run") + assert cost.recorded_cost_usd is None + assert cost.unpriced_tokens == terminal.tokens > 0 + estimate = attribute(trace, "run", rates=RateCard(default=2.0)) + assert estimate.recorded_cost_usd is None + assert estimate.estimated_cost_usd == pytest.approx(terminal.tokens / 1000 * 2) diff --git a/tests/test_slack_recent.py b/tests/test_slack_recent.py new file mode 100644 index 0000000..6e39162 --- /dev/null +++ b/tests/test_slack_recent.py @@ -0,0 +1,139 @@ +"""Bare Slack reader commands select and disclose a confined recent run.""" + +from __future__ import annotations + +import os + +import pytest + +from grapharc.observe.trace import TraceRecorder +from grapharc.slack.bot import handle_text, handle_text_live +from grapharc.slack.config import SlackBotConfig + + +def trace_at(root, directory, run_id): + trace = TraceRecorder(root / "slack-runs" / directory / "trace.jsonl") + trace.event(run_id=run_id, graph="g", node="work", phase="start", step=1) + trace.event(run_id=run_id, graph="g", node="work", phase="end", step=1, tokens=7) + return trace.path + + +@pytest.mark.parametrize("command", ["metrics", "viz", "replay", "trace"]) +def test_bare_reader_names_the_selected_run(tmp_path, command): + path = trace_at(tmp_path, "one", "r1") + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + reply = handle_text(command, config) + assert f"{command} for `slack-runs/one/trace.jsonl`" in reply + assert "run `r1`" in reply + assert "workdir fallback" in reply + assert "arguments are required" not in reply + assert path.read_text() + + +def test_demo_then_bare_metrics_works_without_copying_generated_identifiers(tmp_path): + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + assert "did its job" in handle_text("demo stage0", config) + reply = handle_text("metrics", config) + assert "metrics for `slack-runs/" in reply + assert "run `" in reply + assert "nodes_executed" in reply + + +@pytest.mark.parametrize("command", ["metrics", "viz", "replay", "trace"]) +def test_no_prior_run_is_actionable_not_an_argparse_dump(tmp_path, command): + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + reply = handle_text(command, config) + assert "No prior Slack run" in reply + assert "demo stage0" in reply + assert "arguments are required" not in reply + + +def test_explicit_reader_arguments_are_unchanged(tmp_path): + trace_at(tmp_path, "one", "r1") + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + reply = handle_text("metrics slack-runs/one/trace.jsonl r1", config) + assert "metrics for" not in reply + assert "nodes_executed" in reply + + +def test_latest_file_wins_for_a_workdir_fallback(tmp_path): + old = trace_at(tmp_path, "old", "old-run") + new = trace_at(tmp_path, "new", "new-run") + os.utime(old, ns=(1, 1)) + os.utime(new, ns=(2, 2)) + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + reply = handle_text("metrics", config) + assert "run `new-run`" in reply + + +def test_ambiguous_latest_trace_requires_explicit_selection(tmp_path): + path = trace_at(tmp_path, "two", "r1") + TraceRecorder(path).event(run_id="r2", graph="g", node="n", phase="end", step=1) + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + reply = handle_text("metrics", config) + assert "contains 2 runs" in reply + assert "explicit path and run id" in reply + + +def test_recent_runs_are_confined_before_reading(tmp_path): + outside = tmp_path / "outside" + outside.mkdir() + workdir = tmp_path / "work" + workdir.mkdir() + trace_at(outside, "secret", "private") + (workdir / "slack-runs").symlink_to(outside / "slack-runs", target_is_directory=True) + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=workdir) + assert "escapes" in handle_text("metrics", config) + + +def test_channel_history_precedes_workdir_history_and_survives_missing_paths(tmp_path): + from grapharc.slack.recent import RecentRuns + + history = RecentRuns() + one = trace_at(tmp_path, "one", "r1") + two = trace_at(tmp_path, "two", "r2") + history.remember(tmp_path, "C1", one) + history.remember(tmp_path, "C2", two) + for channel, run_id in [("C1", "r1"), ("C2", "r2")]: + argv, prefix = history.resolve(["metrics"], tmp_path, channel) + assert argv[-1] == run_id + assert "this channel" in prefix + one.unlink() + argv, prefix = history.resolve(["metrics"], tmp_path, "C1") + assert argv[-1] == "r2" + assert "fallback" in prefix + + +def test_live_handler_remembers_the_channel_after_running(tmp_path): + from grapharc.slack.recent import RecentRuns + + history = RecentRuns() + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + handle_text_live("demo stage0", config, None, history=history, channel="C1") + reply = handle_text_live("metrics", config, None, history=history, channel="C1") + assert "this channel" in reply + assert "nodes_executed" in reply + + +def test_corrupt_latest_trace_is_reported_instead_of_answering_about_an_older_run(tmp_path): + old = trace_at(tmp_path, "old", "r1") + new = trace_at(tmp_path, "new", "r2") + os.utime(old, ns=(1, 1)) + new.write_bytes(b"\xff\n") + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=tmp_path) + reply = handle_text("metrics", config) + assert "cannot read the latest Slack trace" in reply + assert "slack-runs/new/trace.jsonl" in reply + + +def test_escaping_candidate_is_skipped_before_it_is_read(tmp_path): + work = tmp_path / "work" + work.mkdir() + trace_at(work, "one", "r1") + outside = tmp_path / "private.jsonl" + outside.write_bytes(b"\xff\n") + link = work / "slack-runs" / "evil" + link.mkdir() + (link / "trace.jsonl").symlink_to(outside) + config = SlackBotConfig(bot_token="unused", app_token="unused", workdir=work) + assert "run `r1`" in handle_text("metrics", config) diff --git a/tests/test_tenant_cost.py b/tests/test_tenant_cost.py index e4c3e7e..4d0123c 100644 --- a/tests/test_tenant_cost.py +++ b/tests/test_tenant_cost.py @@ -100,8 +100,9 @@ def test_error_events_keep_the_tenant_and_spend(tmp_path): cost = tenant_cost(trace, "alpha") assert cost.runs[0].errors == 1 assert cost.recorded_cost_usd == attribute(trace, "failed").recorded_cost_usd + assert cost.recorded_cost_usd == pytest.approx(0.002) assert cost.tokens > 0 - assert cost.unpriced_tokens == cost.tokens + assert cost.unpriced_tokens == 0 def test_tenants_partition_runs_without_double_counting_substeps(tmp_path): @@ -308,9 +309,10 @@ def test_saved_plan_execution_uses_the_current_explicit_tenant( assert {e.tenant for e in events if e.run_id in old_runs} == {"planning"} -def test_governed_planning_and_rounds_share_one_tenant_bill(tmp_path): +@pytest.mark.parametrize("price", [None, 0.003]) +def test_governed_planning_and_rounds_share_one_tenant_bill(tmp_path, price): trace = TraceRecorder(tmp_path / "trace.jsonl") - model = ScriptedChatModel(responses=plan_research.scripted_planner_replies(), cost_usd=0.003) + model = ScriptedChatModel(responses=plan_research.scripted_planner_replies(), cost_usd=price) loop = plan_research.build_loop(model, trace=trace) result = loop.run("explain elevated latency", run_id="loop", tenant="research") events = trace.read_events() @@ -328,14 +330,18 @@ def test_governed_planning_and_rounds_share_one_tenant_bill(tmp_path): assert any(r.executed for r in result.rounds) cost = tenant_cost(trace, "research", rates=RateCard(default=2.0)) assert [r.run_id for r in cost.runs] == ["loop"] - # Planner events currently carry tokens, not a provider price. That - # missing price must remain an estimate, even when the model has one. - assert cost.recorded_cost_usd is None assert cost.tokens == attribute(trace, "loop").tokens > 0 - assert cost.estimated_cost_usd == pytest.approx(cost.tokens * 2.0 / 1000) unpriced = tenant_cost(trace, "research") - assert unpriced.unpriced_tokens == unpriced.tokens - assert unpriced.complete is False + if price is None: + assert cost.recorded_cost_usd is None + assert cost.estimated_cost_usd == pytest.approx(cost.tokens * 2.0 / 1000) + assert unpriced.unpriced_tokens == unpriced.tokens + assert unpriced.complete is False + else: + assert cost.recorded_cost_usd == pytest.approx(model.call_count * price) + assert cost.estimated_cost_usd == 0.0 # only zero-token deterministic nodes + assert unpriced.unpriced_tokens == 0 + assert unpriced.complete is True def test_unlabelled_planner_spend_is_also_priced_or_marked_incomplete(tmp_path):