From 30d730921b58ce39c601440572d2966e1f4dd576 Mon Sep 17 00:00:00 2001 From: blockchainluffy Date: Tue, 27 Oct 2026 14:27:22 +0530 Subject: [PATCH 1/3] fix(password-auth): compare refusal headers without the date header Co-Authored-By: Claude Opus 5.5 --- src/password-auth/authentication.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/password-auth/authentication.test.ts b/src/password-auth/authentication.test.ts index f3eac0a..aaaae6d 100644 --- a/src/password-auth/authentication.test.ts +++ b/src/password-auth/authentication.test.ts @@ -290,8 +290,10 @@ describe("refusing a request", () => { for (const ownRefusal of refusals.slice(8)) { assert.equal(ownRefusal.headers["www-authenticate"], undefined); } - assert.deepEqual(refusals[9]?.headers, refusals[8]?.headers); - assert.deepEqual(refusals[10]?.headers, refusals[8]?.headers); + // Logins using scrypt may finish in different seconds, so compare headers without `date`. + const undated = ({ date: _, ...headers }: Record = {}) => headers; + assert.deepEqual(undated(refusals[9]?.headers), undated(refusals[8]?.headers)); + assert.deepEqual(undated(refusals[10]?.headers), undated(refusals[8]?.headers)); assert.equal((await bearing(issued.token)).statusCode, 200); }); From e98372a5b37b15a32ccf53549de813baa266a6ae Mon Sep 17 00:00:00 2001 From: blockchainluffy Date: Tue, 27 Oct 2026 14:27:57 +0530 Subject: [PATCH 2/3] feat(signals): record model usage per user and stop runs past a usage limit Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 4 +- CONTEXT.md | 15 +- README.md | 1 + examples/05_usage_limits/.dockerignore | 3 + examples/05_usage_limits/.env.example | 6 + examples/05_usage_limits/AGENTS.md | 25 ++ examples/05_usage_limits/Dockerfile | 10 + examples/05_usage_limits/Dockerfile.agent | 8 + examples/05_usage_limits/README.md | 110 ++++++++ examples/05_usage_limits/compose.yml | 112 +++++++++ examples/05_usage_limits/drizzle.config.ts | 18 ++ examples/05_usage_limits/main.ts | 114 +++++++++ examples/05_usage_limits/package.json | 23 ++ examples/05_usage_limits/schema.ts | 4 + examples/05_usage_limits/settings.json | 4 + examples/05_usage_limits/tsconfig.json | 19 ++ scripts/check-package.ts | 51 +++- site/architecture.md | 3 +- src/gateway/gateway.test.ts | 19 +- src/gateway/gateway.ts | 16 +- src/pi/agent-instance.test.ts | 69 +++++- src/pi/output.test.ts | 276 ++++++++++++++++++++- src/pi/output.ts | 94 ++++++- src/pi/rpc.ts | 7 + src/pi/runtime.test.ts | 99 +++++++- src/pi/runtime.ts | 33 ++- src/signals/handlers.ts | 21 +- src/signals/index.ts | 4 +- src/signals/runtime.ts | 41 ++- src/signals/schema/index.ts | 34 ++- src/signals/template-handler.test.ts | 51 +++- src/signals/template-handler.ts | 7 +- src/signals/usage.test.ts | 183 ++++++++++++++ src/signals/usage.ts | 97 ++++++++ src/signals/worker.test.ts | 230 ++++++++++++++++- src/signals/worker.ts | 48 ++-- src/test-support/fake-runtime.ts | 11 +- 37 files changed, 1804 insertions(+), 66 deletions(-) create mode 100644 examples/05_usage_limits/.dockerignore create mode 100644 examples/05_usage_limits/.env.example create mode 100644 examples/05_usage_limits/AGENTS.md create mode 100644 examples/05_usage_limits/Dockerfile create mode 100644 examples/05_usage_limits/Dockerfile.agent create mode 100644 examples/05_usage_limits/README.md create mode 100644 examples/05_usage_limits/compose.yml create mode 100644 examples/05_usage_limits/drizzle.config.ts create mode 100644 examples/05_usage_limits/main.ts create mode 100644 examples/05_usage_limits/package.json create mode 100644 examples/05_usage_limits/schema.ts create mode 100644 examples/05_usage_limits/settings.json create mode 100644 examples/05_usage_limits/tsconfig.json create mode 100644 src/signals/usage.test.ts create mode 100644 src/signals/usage.ts diff --git a/CLAUDE.md b/CLAUDE.md index bb4c957..7b7ffa2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -36,7 +36,7 @@ Docker or the network: - **`npm run check:package`** packs the tarball, installs it into a throwaway project, and proves every `exports` subpath resolves there for Node *and* the type checker. It is also the only thing - anywhere that holds the thirteen table names apart, and the only proof that the root still + anywhere that holds the fourteen table names apart, and the only proof that the root still resolves to nothing. - **`npm run test:container`** is the single end-to-end test: a real container running real `pi` against the real Agent server, with a scripted model on localhost, so it needs no model @@ -114,7 +114,7 @@ Four ways that fails quietly, all of them a deployment's to get right: schema module declares carries a component prefix, because **`export *` drops a name that resolves to more than one binding**: eight components exporting a bare `schema` yield a barrel exporting none, `schemaFilter` derives to one schema, and `push` prints `Changes applied` having created one - component's tables. Two names are in that class and no more; the thirteen table names and the four + component's tables. Two names are in that class and no more; the fourteen table names and the four enum arrays are not prefixed. - **The table names are guarded by `scripts/check-package.ts`, which imports all eight `/schema` specifiers un-aliased into one module scope.** A duplicate is a `TS2300` and a `SyntaxError` diff --git a/CONTEXT.md b/CONTEXT.md index 4bb8d6e..d7e3504 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -129,7 +129,7 @@ Operator-defined JSON a Signal Handler writes on a Prompt beside the Session and _Avoid_: scope, claims, permissions, grant, ACL **Post phase**: -The Signal Handler's second entry point, for cleanup. Runs once after all Runs arising from a Signal have finished, receives a flag saying whether one failed, and cannot produce Prompts. +The Signal Handler's second entry point, used for cleanup. It runs once after all Runs from a Signal finish. It receives two flags: whether any Run failed and whether any Run exceeded its Usage limit. It cannot produce Prompts. _Avoid_: post handler, teardown, finalizer, callback **Session**: @@ -140,6 +140,19 @@ _Avoid_: chat, conversation, thread, context (as a name for a Session; a Prompt' One execution of the agent: a single Prompt, in one Session, producing whatever the agent emits. _Avoid_: turn, job, invocation, task +**Usage**: +The model resources a Run uses, and the always-on infrastructure Component that reads the recorded usage. The Runtime reports usage for each model call, including retries and calls that compact the Session's history. The Signal Worker immediately saves each report as a **Usage record** linked to its Run. Saved records survive restarts. If a record cannot be saved, the Run stops to prevent further unrecorded usage. + +The Agent Implementation copies the provider's token counts. Token totals do not directly measure cost because providers price each token category differently. The framework can read totals for a time period, but does not enforce a budget from those totals. The Operator defines the budget, its time period, and messages to Users in a Signal Handler. The name "Usage" avoids confusion with "Token", which means a credential issued to a User. +_Avoid_: Token Tracker, billing, metering, quota + +**Usage record**: +One saved row for one model call in a Run. It contains the provider, model, four token counts, their sum, and the cost calculated by the Agent Implementation. Records are written once and never updated. + +**Usage limit**: +The maximum model token count for one Run, set by a Signal Handler on the Prompt. The Runtime adds each call's usage when the call finishes. If the total exceeds the limit, it stops the agent and marks the Run as failed and limited. The Session keeps its history up to that point. A Run can exceed the limit by one call because usage is only known after a call finishes. This per-Run limit is the only limit the framework enforces. To enforce a budget, a Signal Handler passes the remaining token allowance. The limit counts every token category at full weight, including cache reads. Each model call sends the agent's whole context again and the provider reports most of it as cache reads, so cache reads usually make up most of a Run's count, although providers charge much less for them. +_Avoid_: budget, quota, max tokens + **Workspace**: The files and data that Signal Handlers and the agent share, as opposed to the Db, which the agent cannot touch directly. Global to a shared agent, not per Session. **The framework has no name for where it is.** The Gateway and the **Agent Instance** come to share it by the Operator's own arrangement, a volume or a bind or a filesystem both containers happen to see, and no option, no path and no check anywhere in the API is about it: nothing here can tell whether the two ends found each other, or notice that they did not. _Avoid_: scratch, working directory, shared state diff --git a/README.md b/README.md index 7b8eb37..709ff8c 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,7 @@ from the registry rather than this checkout. | [Decisions](./examples/02_decisions/) | Two users and a shared record of signed decisions | | [Nostr](./examples/03_nostr/) | Encrypted direct messages through a local relay | | [Agent isolation](./examples/04_agent_isolation/) | Restricted agent networking and a model proxy that holds the credential | +| [Usage limits](./examples/05_usage_limits/) | Two users, each with a daily model usage limit | ## Deployment boundaries diff --git a/examples/05_usage_limits/.dockerignore b/examples/05_usage_limits/.dockerignore new file mode 100644 index 0000000..c7234c5 --- /dev/null +++ b/examples/05_usage_limits/.dockerignore @@ -0,0 +1,3 @@ +node_modules +state +.env diff --git a/examples/05_usage_limits/.env.example b/examples/05_usage_limits/.env.example new file mode 100644 index 0000000..7be2243 --- /dev/null +++ b/examples/05_usage_limits/.env.example @@ -0,0 +1,6 @@ +ANTHROPIC_API_KEY= + +ALICE_PASSWORD=correct horse battery staple +BOB_PASSWORD=chased by a dog on a bicycle + +DAILY_TOKEN_LIMIT=20000 diff --git a/examples/05_usage_limits/AGENTS.md b/examples/05_usage_limits/AGENTS.md new file mode 100644 index 0000000..7aa72bc --- /dev/null +++ b/examples/05_usage_limits/AGENTS.md @@ -0,0 +1,25 @@ +# Shared agent instructions + +You help one user through a gateway. Keep answers brief. + +## Gateway API + +Use your shell and `curl` to reach `$AGENT_SERVER_URL`. No credential is required. +Read the live API description before calling it: + +```sh +curl -s "$AGENT_SERVER_URL/openapi.json" +``` + +Use that document for request fields and error responses. If a call is refused, check the +response and correct the request before retrying. Do not repeatedly send the same invalid call. + +## Reply to the user + +Send replies with `POST /messages` and body `{"userId":"...","text":"..."}`. +Copy `userId` from the triggering signal; never invent one. Your final model response and +workspace files are not delivered to the user. + +Read prior messages with `GET /messages?user=`. The message log contains both directions; +use its cursors when paging. Write short, plain sentences without headings, tables, or code +blocks. The terminal client polls once a second, so delivery may follow your run. diff --git a/examples/05_usage_limits/Dockerfile b/examples/05_usage_limits/Dockerfile new file mode 100644 index 0000000..29d0336 --- /dev/null +++ b/examples/05_usage_limits/Dockerfile @@ -0,0 +1,10 @@ +FROM node:24-alpine + +WORKDIR /app + +COPY package.json ./ +RUN npm install --no-audit --no-fund + +COPY main.ts drizzle.config.ts schema.ts ./ + +CMD ["node", "main.ts"] diff --git a/examples/05_usage_limits/Dockerfile.agent b/examples/05_usage_limits/Dockerfile.agent new file mode 100644 index 0000000..26211d5 --- /dev/null +++ b/examples/05_usage_limits/Dockerfile.agent @@ -0,0 +1,8 @@ +FROM node:24-alpine + +RUN apk add --no-cache curl socat + +RUN npm install -g @earendil-works/pi-coding-agent@0.85.1 + +WORKDIR /workspace +ENV PI_CODING_AGENT_DIR=/home/agent/.pi/agent diff --git a/examples/05_usage_limits/README.md b/examples/05_usage_limits/README.md new file mode 100644 index 0000000..de9e83b --- /dev/null +++ b/examples/05_usage_limits/README.md @@ -0,0 +1,110 @@ +# Usage limits example + +Alice and Bob talk to one agent over HTTP. Each user has a daily model token limit. When Alice +reaches her limit, the gateway replies to her next message without calling the model. Bob can +still get answers. This example defines the daily limit in `main.ts`. The framework records +Usage and stops individual Runs when they exceed their Usage limit. + +## Requirements + +This example requires Usage support in `@shutter-network/concorde`, so `package.json` specifies +`^0.2.0`. Until that release is published to npm, `npm install` and `docker compose up --build` +will fail. The example installs Concorde from the npm registry, not from this checkout. + +## Run + +You need Docker with Compose and an Anthropic API key. From this directory: + +```sh +cp .env.example .env +``` + +Set `ANTHROPIC_API_KEY`, `ALICE_PASSWORD`, and `BOB_PASSWORD` in `.env`. `DAILY_TOKEN_LIMIT` is +the number of model tokens each user may use per UTC day. The value in `.env.example` is small +so you can reach the limit with a few messages. Check the model in `settings.json`. + +```sh +docker compose up -d --build +docker compose logs -f gateway +``` + +The log lists each user's id and name. Open the two clients in separate terminals: + +```sh +docker compose run --rm tui-alice +docker compose run --rm tui-bob +``` + +Send messages as Alice until the agent stops answering and she receives "Your daily usage limit +is reached. It resets at 00:00 UTC." Every further message from her that day gets the same +answer from the gateway, and the model is not called. Then send a message as Bob: his Usage is +counted separately, so he still gets answers. + +The public API documentation is at . + +## How it works + +`main.ts` uses a custom Signal Handler for `messageReceivedKind`. The handler checks Usage +before deciding whether to prompt the agent, so it does not use `templateHandler`. + +- `handle` calls `usage.total` to count the sender's tokens since 00:00 UTC today. If the total + is at or above `DAILY_TOKEN_LIMIT`, it sends the limit message through the Messenger and + returns no Prompt, so no Run starts. Otherwise, it returns one Prompt with the sender's + `userId` and sets `usageLimit` to their remaining daily token allowance. The Run's Usage is + added to that user's total. +- `post` receives the Run's outcome. If `limited` is true, the Run exceeded its Usage limit + and the handler sends the limit message. For any other failure, it sends "I could not complete + that request. Please try again in a minute." + +The Prompt's `userId` comes from the Message authenticated by the Gateway. It never comes from +text written by a model or sender. The Session name includes the user id, so later Runs continue +that user's Session. Stopping a Run at its Usage limit preserves the Session. The user's next +allowed message on the following day continues the same Session. + +On the first start, trusted application code creates both users with their names and passwords +in one transaction. Restarts keep the saved users and passwords. Changing passwords in `.env` +does not change the saved passwords, even though the example prints the `.env` values again. + +## Limit behavior and requirements + +- **Usage can exceed the limit by one model response.** Usage is counted when each response + ends. The Run stops once its total exceeds `usageLimit`, but the provider still charges for + that response. A user's daily total can therefore exceed `DAILY_TOKEN_LIMIT`. The gateway + replies to their next message without calling the model. +- **The model server must report usage.** The counts are the ones the model provider returns + with each response, copied by `pi`. Some self-hosted and OpenAI-compatible servers report no + usage. Those responses are recorded as 0 tokens, so the limit never takes effect. If you + configure another provider in `models.json`, check that it reports usage before relying on + the limit. Responses interrupted mid-stream can also report fewer tokens than were used. +- **The counts are tokens, not money.** `DAILY_TOKEN_LIMIT` is compared with the total of input, + output, cache read, and cache write tokens. Providers price those categories differently, so + equal token totals can cost different amounts. +- **Most of the count is cache read tokens.** Every model call sends the agent's whole context + again: the system prompt, `AGENTS.md`, the tool definitions, and the conversation so far. The + provider serves most of that context from its cache and reports it as cache read tokens. A + reply is usually a few hundred output tokens, but one message can count 4,000 to 10,000 tokens + in total, and the count per message grows as the conversation gets longer. Providers usually + charge about a tenth of the normal input price for cache reads, but `DAILY_TOKEN_LIMIT` counts + them at full weight. Choose the limit with that in mind. To count the categories differently, + calculate your own number in the Signal Handler from the fields that `usage.total()` returns, + instead of using `totalTokens`. +- **Each user has a separate limit.** Runs without a `userId` do not count toward any user's + total. To set a hard spending cap for the whole deployment, enforce it at the model provider + or a model proxy that holds the credential, as in + [the isolation example](../04_agent_isolation/). + +## Files and storage + +- `main.ts`: component setup, the daily limit handler, user creation, and shutdown. +- `AGENTS.md`: mounted agent instructions; `settings.json`: model configuration. +- `compose.yml` and the Dockerfiles: gateway, pi RPC listener, database, migration, and two + clients. +- `schema.ts`: tables for the four components. Usage records are part of the Signals schema, + so they need no separate schema setup. +- `state/workspace/` and `state/sessions/`: persistent agent files and session records. + +Keep the agent server and RPC listener private. The [guide](../../site/guide.md) explains the +shared deployment setup. + +Stop with `docker compose down`. Add `-v` to delete the database volume, which also deletes the +recorded Usage; local `state/` files remain. diff --git a/examples/05_usage_limits/compose.yml b/examples/05_usage_limits/compose.yml new file mode 100644 index 0000000..b6a24e4 --- /dev/null +++ b/examples/05_usage_limits/compose.yml @@ -0,0 +1,112 @@ +name: concorde-example-05-usage-limits + +x-database-url: &database-url postgres://concorde:concorde@postgres:5432/concorde + +x-tui: &tui + build: + context: . + dockerfile: Dockerfile + profiles: [cli] + entrypoint: ["npx", "concorde-chat-tui"] + networks: [public] + +services: + gateway: + build: + context: . + dockerfile: Dockerfile + environment: + DATABASE_URL: *database-url + ALICE_PASSWORD: ${ALICE_PASSWORD:?copy .env.example to .env} + BOB_PASSWORD: ${BOB_PASSWORD:?copy .env.example to .env} + DAILY_TOKEN_LIMIT: ${DAILY_TOKEN_LIMIT:?copy .env.example to .env} + PUBLIC_HOST: 0.0.0.0 + PUBLIC_PORT: "8085" + AGENT_SERVER_HOST: 0.0.0.0 + AGENT_SERVER_PORT: "7411" + AGENT_INSTANCE_HOST: agent + AGENT_INSTANCE_PORT: "4000" + AGENT_SESSIONS_DIR: /sessions + ports: + - "127.0.0.1:8085:8085" + networks: [db, agent, public] + depends_on: + postgres: + condition: service_healthy + migrate: + condition: service_completed_successfully + agent: + condition: service_started + stop_grace_period: 300s + + agent: + build: + context: . + dockerfile: Dockerfile.agent + command: + - socat + - TCP-LISTEN:4000,reuseaddr,fork + - EXEC:pi --mode rpc --no-approve + environment: + ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY:?copy .env.example to .env and put your model key in it} + AGENT_SERVER_URL: http://gateway:7411 + PI_OFFLINE: "1" + PI_CODING_AGENT_DIR: /home/agent/.pi/agent + volumes: + - ./state/workspace:/workspace + - ./state/agent:/home/agent/.pi/agent + - ./state/sessions:/sessions + - ./AGENTS.md:/workspace/AGENTS.md:ro + - ./settings.json:/home/agent/.pi/agent/settings.json:ro + networks: [agent] + + migrate: + build: + context: . + dockerfile: Dockerfile + command: ["npx", "drizzle-kit", "push", "--force"] + environment: + DATABASE_URL: *database-url + restart: "no" + networks: [db] + depends_on: + postgres: + condition: service_healthy + + postgres: + image: postgres:17 + environment: + POSTGRES_USER: concorde + POSTGRES_PASSWORD: concorde + POSTGRES_DB: concorde + volumes: + - db:/var/lib/postgresql/data + networks: [db] + healthcheck: + test: ["CMD", "pg_isready", "-U", "concorde", "-d", "concorde"] + interval: 3s + timeout: 3s + retries: 20 + + tui-alice: + <<: *tui + environment: + CONCORDE_GATEWAY_URL: http://gateway:8085 + CONCORDE_PASSWORD: ${ALICE_PASSWORD:?copy .env.example to .env} + + tui-bob: + <<: *tui + environment: + CONCORDE_GATEWAY_URL: http://gateway:8085 + CONCORDE_PASSWORD: ${BOB_PASSWORD:?copy .env.example to .env} + +networks: + db: + name: concorde_usage_limits_db + agent: + name: concorde_usage_limits_agent + public: + name: concorde_usage_limits_public + +volumes: + db: diff --git a/examples/05_usage_limits/drizzle.config.ts b/examples/05_usage_limits/drizzle.config.ts new file mode 100644 index 0000000..cd8b8d2 --- /dev/null +++ b/examples/05_usage_limits/drizzle.config.ts @@ -0,0 +1,18 @@ +import { defineConfig } from "drizzle-kit"; +import { is } from "drizzle-orm"; +import { PgSchema } from "drizzle-orm/pg-core"; +import * as schema from "./schema.ts"; + +const databaseUrl = process.env.DATABASE_URL; +if (databaseUrl === undefined) { + throw new Error("Set DATABASE_URL to the PostgreSQL connection URL."); +} + +export default defineConfig({ + dialect: "postgresql", + schema: "./schema.ts", + schemaFilter: Object.values(schema) + .filter((exported) => is(exported, PgSchema)) + .map((pgSchema) => pgSchema.schemaName), + dbCredentials: { url: databaseUrl }, +}); diff --git a/examples/05_usage_limits/main.ts b/examples/05_usage_limits/main.ts new file mode 100644 index 0000000..c61a064 --- /dev/null +++ b/examples/05_usage_limits/main.ts @@ -0,0 +1,114 @@ +import { createGateway } from "@shutter-network/concorde/gateway"; +import { createHttpChannel } from "@shutter-network/concorde/http-channel"; +import { + createMessenger, + type MessageRecord, + messageReceivedKind, +} from "@shutter-network/concorde/messenger"; +import { createPasswordAuth } from "@shutter-network/concorde/password-auth"; +import { createPiRuntime } from "@shutter-network/concorde/pi"; +import type { SignalHandler } from "@shutter-network/concorde/signals"; +import { createUsers } from "@shutter-network/concorde/users"; + +const people = [ + { name: "alice", password: process.env.ALICE_PASSWORD! }, + { name: "bob", password: process.env.BOB_PASSWORD! }, +]; + +const dailyTokenLimit = Number(process.env.DAILY_TOKEN_LIMIT); +if (!Number.isSafeInteger(dailyTokenLimit) || dailyTokenLimit <= 0) { + throw new Error("Set DAILY_TOKEN_LIMIT to a positive whole number of tokens."); +} + +const limitReached = "Your daily usage limit is reached. It resets at 00:00 UTC."; +const tryAgain = "I could not complete that request. Please try again in a minute."; + +const tokenTtl = 30 * 24 * 60 * 60 * 1000; + +function startOfUtcDay(now: Date): Date { + return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate())); +} + +const runtime = createPiRuntime({ + host: process.env.AGENT_INSTANCE_HOST!, + port: Number(process.env.AGENT_INSTANCE_PORT), + sessionsDir: process.env.AGENT_SESSIONS_DIR!, +}); + +const gateway = createGateway({ + databaseUrl: process.env.DATABASE_URL!, + runtime, + publicListen: { host: process.env.PUBLIC_HOST!, port: Number(process.env.PUBLIC_PORT) }, + agentListen: { + host: process.env.AGENT_SERVER_HOST!, + port: Number(process.env.AGENT_SERVER_PORT), + }, + extend: ({ db, agentServer, publicServer, worker }) => { + const users = createUsers({ db, agentServer, publicServer }); + const passwordAuth = createPasswordAuth({ db, users, publicServer, tokenTtl }); + const messenger = createMessenger({ db, users, worker, agentServer }); + const httpChannel = createHttpChannel({ db, messenger, publicServer }); + return { users, passwordAuth, messenger, httpChannel }; + }, + handlers: ({ db, usage, messenger }) => { + const tell = (userId: string, text: string) => db.tx((tx) => messenger.send(tx, userId, text)); + + const messageHandler: SignalHandler = { + async handle(signal) { + const { userId, text } = signal.payload; + const used = await usage.total({ userId, since: startOfUtcDay(new Date()) }); + if (used.totalTokens >= dailyTokenLimit) { + await tell(userId, limitReached); + return []; + } + return [ + { + session: `user_${userId}`, + text: `User ${userId} sent this message: + +${text} + +Send your reply through the gateway messaging API. Your final model response is not delivered.`, + userId, + usageLimit: dailyTokenLimit - used.totalTokens, + }, + ]; + }, + async post(signal, outcome) { + if (outcome.limited) { + await tell(signal.payload.userId, limitReached); + } else if (outcome.failed) { + await tell(signal.payload.userId, tryAgain); + } + }, + }; + + return { [messageReceivedKind]: messageHandler }; + }, +}); + +await gateway.start(); + +const { db, users, passwordAuth } = gateway.components; + +if ((await users.list({ limit: 1 })).length === 0) { + await db.tx(async (tx) => { + for (const person of people) { + const user = await users.create(tx); + await users.setAttributes(tx, user.id, { name: person.name }); + await passwordAuth.setPassword(tx, user.id, person.password); + } + }); +} + +for (const user of await users.list()) { + console.log(`user ${user.id} ${JSON.stringify(user.attributes)}`); +} +for (const person of people) { + console.log(`${person.name} logs in with the password ${person.password}`); +} +console.log(`each user may use ${dailyTokenLimit} model tokens per UTC day`); + +for (const stopping of ["SIGINT", "SIGTERM"] as const) { + process.once(stopping, () => void gateway.stop()); +} diff --git a/examples/05_usage_limits/package.json b/examples/05_usage_limits/package.json new file mode 100644 index 0000000..db783e6 --- /dev/null +++ b/examples/05_usage_limits/package.json @@ -0,0 +1,23 @@ +{ + "name": "concorde-example-05-usage-limits", + "private": true, + "type": "module", + "engines": { + "node": ">=24" + }, + "scripts": { + "typecheck": "tsc --noEmit", + "migrate": "drizzle-kit push --force", + "start": "node main.ts" + }, + "dependencies": { + "drizzle-orm": "^0.45.2", + "fastify": "^5.12.5", + "@shutter-network/concorde": "^0.2.0" + }, + "devDependencies": { + "@types/node": "^24.13.3", + "drizzle-kit": "^0.31.10", + "typescript": "^7.0.2" + } +} diff --git a/examples/05_usage_limits/schema.ts b/examples/05_usage_limits/schema.ts new file mode 100644 index 0000000..0a5bd37 --- /dev/null +++ b/examples/05_usage_limits/schema.ts @@ -0,0 +1,4 @@ +export * from "@shutter-network/concorde/messenger/schema"; +export * from "@shutter-network/concorde/password-auth/schema"; +export * from "@shutter-network/concorde/signals/schema"; +export * from "@shutter-network/concorde/users/schema"; diff --git a/examples/05_usage_limits/settings.json b/examples/05_usage_limits/settings.json new file mode 100644 index 0000000..63632ef --- /dev/null +++ b/examples/05_usage_limits/settings.json @@ -0,0 +1,4 @@ +{ + "defaultModel": "claude-sonnet-5", + "defaultProvider": "anthropic" +} diff --git a/examples/05_usage_limits/tsconfig.json b/examples/05_usage_limits/tsconfig.json new file mode 100644 index 0000000..cdea019 --- /dev/null +++ b/examples/05_usage_limits/tsconfig.json @@ -0,0 +1,19 @@ +{ + "compilerOptions": { + "target": "es2024", + "lib": ["es2024"], + "module": "nodenext", + "moduleResolution": "nodenext", + "types": ["node"], + + "strict": true, + "verbatimModuleSyntax": true, + + "allowImportingTsExtensions": true, + "erasableSyntaxOnly": true, + + "skipLibCheck": true, + "noEmit": true + }, + "include": ["main.ts", "drizzle.config.ts", "schema.ts"] +} diff --git a/scripts/check-package.ts b/scripts/check-package.ts index 8ef37ef..e666f6d 100644 --- a/scripts/check-package.ts +++ b/scripts/check-package.ts @@ -19,8 +19,8 @@ const consumerImports = [ 'import { openDb } from "@shutter-network/concorde/db";', 'import { createBareGateway, createGateway, NoAuthRegisteredError, serverComponent } from "@shutter-network/concorde/gateway";', 'import { defaultLogger } from "@shutter-network/concorde/logging";', - 'import { createSignalWorker, runStates, signalStates, templateHandler } from "@shutter-network/concorde/signals";', - 'import { runs, signals, signalsSchema, signalsTables } from "@shutter-network/concorde/signals/schema";', + 'import { createSignalWorker, createUsage, runStates, signalStates, templateHandler } from "@shutter-network/concorde/signals";', + 'import { runs, signals, signalsSchema, signalsTables, usage } from "@shutter-network/concorde/signals/schema";', 'import { createPiRuntime } from "@shutter-network/concorde/pi";', 'import { createUsers } from "@shutter-network/concorde/users";', 'import { users, usersSchema, usersTables } from "@shutter-network/concorde/users/schema";', @@ -127,6 +127,8 @@ try { "dist/signals/schema/index.d.ts", "dist/signals/routes.js", "dist/signals/routes.d.ts", + "dist/signals/usage.js", + "dist/signals/usage.d.ts", "dist/users/index.js", "dist/users/index.d.ts", "dist/users/routes.js", @@ -387,6 +389,7 @@ try { " EmittedSignal,", " PostOutcome,", " Prompt,", + " RunObserver,", " RunOutcome,", " RunPrompt,", " RunRecord,", @@ -400,6 +403,11 @@ try { " SignalWorker,", " SignalWorkerOptions,", " TemplateHandlerOptions,", + " Usage,", + " UsageOptions,", + " UsageQuery,", + " UsageReport,", + " UsageTotals,", '} from "@shutter-network/concorde/signals";', "import type {", " PiInstance,", @@ -479,10 +487,10 @@ try { "// wrapper would be dropped in silence and generate an empty migration. Annotated as the two", "// Drizzle types the exporter actually filters on, so a component that stopped exporting a", "// table fails here rather than at an Operator's first `generate`. Nothing here is aliased:", - "// thirteen table names in one scope is a deployment's own barrel, and a fourteenth colliding", + "// fourteen table names in one scope is a deployment's own barrel, and a fifteenth colliding", "// with one of them is a compile error rather than a table missing from the DDL.", "export const partTables: readonly PgTable[] = [", - " signals, runs, users, passwords, tokens,", + " signals, runs, usage, users, passwords, tokens,", " grants, admitted,", " messages, pubkeys, received, outbox, decisions, schedules,", "];", @@ -519,8 +527,18 @@ try { "// framework base type to import. It takes a `RunPrompt` and not a `Prompt`,", "// which is the seam's own type and the reason `prompt.session` is a string", "// here with nothing to check: the Signal Worker settled it.", + "// The worker always passes an observer, though runtimes may omit the parameter.", + "// Each report describes one model call and is saved by the worker as a Usage record.", "const runtime: Runtime = {", - " async run(prompt: RunPrompt): Promise {", + " async run(prompt: RunPrompt, observer?: RunObserver): Promise {", + " const report: UsageReport = {", + ' provider: "own", model: null, input: 1, output: 1, cacheRead: 0, cacheWrite: 0,', + " totalTokens: 2, cost: 0,", + " };", + " await observer?.usage(report);", + " if (prompt.usageLimit !== undefined && report.totalTokens > prompt.usageLimit) {", + ' return { ok: false, error: "over the limit", limited: true };', + " }", ' return prompt.text === "" ? { ok: false, error: "nothing to say in " + prompt.session } : { ok: true };', " },", "};", @@ -706,10 +724,17 @@ try { "function greeter(greeting: string): SignalHandler<{ userId: string }> {", " return {", " handle: (signal: Signal<{ userId: string }>): Prompt[] => [", - ' { session: "user_" + signal.payload.userId, text: greeting + ", " + signal.id },', + " // Assign usage to the User identified by the Gateway in the Signal.", + " {", + ' session: "user_" + signal.payload.userId,', + ' text: greeting + ", " + signal.id,', + " userId: signal.payload.userId,", + " usageLimit: 20_000,", + " },", " ],", " post: (_signal: Signal<{ userId: string }>, outcome: PostOutcome) => {", - ' if (outcome.failed) log.warn({}, "a Run failed");', + ' if (outcome.limited) log.warn({}, "a Run reached its usage limit");', + ' else if (outcome.failed) log.warn({}, "a Run failed");', " },", " };", "}", @@ -728,6 +753,7 @@ try { ' template: "You were told: {{said}}\\n{{> footer}}",', ' session: (signal: Signal) => "user_" + signal.payload.userId,', " data: (signal: Signal) => ({ said: signal.payload.text }),", + " user: (signal: Signal) => signal.payload.userId,", " helpers: { shout: (value: string) => value.toUpperCase() },", ' partials: { footer: "-- sent by the Gateway" },', " post: (signal: Signal, outcome: PostOutcome) => {", @@ -758,6 +784,15 @@ try { "export const worker: SignalWorker = createSignalWorker(options);", "export const workerComponent: Component = worker;", "", + "// Usage reads the worker's saved records. `createGateway` creates this component for you.", + "// To create it yourself, pass only the Db.", + "const usageOptions: UsageOptions = { db };", + "export const usageReader: Usage = createUsage(usageOptions);", + 'const today: UsageQuery = { userId: "42", since: new Date(0), until: new Date() };', + "export async function spentToday(): Promise {", + " return usageReader.total(today);", + "}", + "", "// The Signal and Run routes stay exported as a plugin even though the Agent server", "// already has them: passing the server is the easy path, and holding the plugin is", "// the door out — a prefix of the consumer's own, or their own encapsulated scope", @@ -1393,7 +1428,7 @@ try { ); assert.equal( imported, - "function:function:sessionsDir must be an absolute path on the agent instance:Session user_42: Cannot connect to agent instance at 127.0.0.1:1:Session ../escape: Invalid session name:run:concorde_users:agentRoutes,create,get,list,setAttributes,start,stop:password auth Bearer authenticate,issueToken,revoke,scheme,setPassword,start,stop in concorde_password_auth:nostr auth Nostr authenticate,recordPublicKey,scheme,start,stop in concorde_nostr_auth:concorde_messenger:history,register,send,start,stop:channel http name,send,start,stop:channel nostr drain,name,publicKey,recordPublicKey,send,start,stop as 1b84c5567b126440995d3ed5aaba0565d71e1834604819ff9c17f5e9d5dd078f in concorde_nostr_channel:message.received:concorde_decisions:sign,start,stop:history,publish,start,stop:3 segments, 64 signature bytes, verified true, private member false:db,agentServer,publicServer,users,passwordAuth,signatures,decisions,messenger,httpChannel,ownLoop,worker:Concorde: agent server describes 10 paths:by hand /healthz:cron 2030-06-02T09:00:00.000Z zone true:scheduler cancel,list,schedule,start,stop,tick fires concorde_schedule_fired in concorde_scheduler:tables concorde_decisions.decisions concorde_messenger.messages concorde_nostr_auth.admitted concorde_nostr_auth.grants concorde_nostr_channel.outbox concorde_nostr_channel.pubkeys concorde_nostr_channel.received concorde_password_auth.passwords concorde_password_auth.tokens concorde_scheduler.schedules concorde_signals.runs concorde_signals.signals concorde_users.users in concorde_decisions concorde_messenger concorde_nostr_auth concorde_nostr_channel concorde_password_auth concorde_scheduler concorde_signals concorde_users, wrappers seen 0, wrappers present 8:schemas 8 of 8 distinct, on a component subpath 0, resolving to files true", + "function:function:sessionsDir must be an absolute path on the agent instance:Session user_42: Cannot connect to agent instance at 127.0.0.1:1:Session ../escape: Invalid session name:run:concorde_users:agentRoutes,create,get,list,setAttributes,start,stop:password auth Bearer authenticate,issueToken,revoke,scheme,setPassword,start,stop in concorde_password_auth:nostr auth Nostr authenticate,recordPublicKey,scheme,start,stop in concorde_nostr_auth:concorde_messenger:history,register,send,start,stop:channel http name,send,start,stop:channel nostr drain,name,publicKey,recordPublicKey,send,start,stop as 1b84c5567b126440995d3ed5aaba0565d71e1834604819ff9c17f5e9d5dd078f in concorde_nostr_channel:message.received:concorde_decisions:sign,start,stop:history,publish,start,stop:3 segments, 64 signature bytes, verified true, private member false:db,agentServer,publicServer,usage,users,passwordAuth,signatures,decisions,messenger,httpChannel,ownLoop,worker:Concorde: agent server describes 10 paths:by hand /healthz:cron 2030-06-02T09:00:00.000Z zone true:scheduler cancel,list,schedule,start,stop,tick fires concorde_schedule_fired in concorde_scheduler:tables concorde_decisions.decisions concorde_messenger.messages concorde_nostr_auth.admitted concorde_nostr_auth.grants concorde_nostr_channel.outbox concorde_nostr_channel.pubkeys concorde_nostr_channel.received concorde_password_auth.passwords concorde_password_auth.tokens concorde_scheduler.schedules concorde_signals.runs concorde_signals.signals concorde_signals.usage concorde_users.users in concorde_decisions concorde_messenger concorde_nostr_auth concorde_nostr_channel concorde_password_auth concorde_scheduler concorde_signals concorde_users, wrappers seen 0, wrappers present 8:schemas 8 of 8 distinct, on a component subpath 0, resolving to files true", "Installed exports, component construction, schema exports, and runtime dependencies must match the public API", ); diff --git a/site/architecture.md b/site/architecture.md index f7b03f3..191a7b0 100644 --- a/site/architecture.md +++ b/site/architecture.md @@ -28,13 +28,14 @@ The diagram shows messaging and authentication options. A messenger accepts one current implementation; the two channels shown are alternatives. The scheduler, signatures, and decisions can also be added, although they are not drawn. -`createGateway` constructs four infrastructure components: +`createGateway` constructs five infrastructure components: | Component | Responsibility | | --- | --- | | Db | Shared PostgreSQL access so component state changes can commit together | | Agent server | HTTP capabilities the agent uses to read application data and act | | Public server | HTTP endpoints for user clients, including authentication and message exchange | +| Usage | Totals of the model usage the signal worker records for each run | | Signal worker | Durable queue that turns signals into agent runs, one signal at a time | Your `extend` callback constructs additional components using this infrastructure. Dependencies diff --git a/src/gateway/gateway.test.ts b/src/gateway/gateway.test.ts index 329b849..7348a58 100644 --- a/src/gateway/gateway.test.ts +++ b/src/gateway/gateway.test.ts @@ -47,12 +47,13 @@ const nobody = "2f1b4d54-1c3a-4f2e-9d7b-8e6a5c4b3a21"; const { privateKey: signingKey } = generateKeyPairSync("ed25519"); -const theInfra = ["db", "agentServer", "publicServer", "worker"]; +const theInfra = ["db", "agentServer", "publicServer", "usage", "worker"]; const theRecord = [ "db", "agentServer", "publicServer", + "usage", "users", "passwordAuth", "signatures", @@ -673,7 +674,7 @@ describe("a whole deployment from one call", () => { }); describe("the infrastructure on its own", () => { - it("is the four infrastructure Components and nothing else when no extend is passed", () => { + it("is the five infrastructure Components and nothing else when no extend is passed", () => { const bare = createGateway({ databaseUrl: "postgres://nobody@example.invalid/none", runtime, @@ -702,7 +703,7 @@ describe("the infrastructure on its own", () => { ); }); - it("refuses an extend that returns one of the four infrastructure keys", () => { + it("refuses an extend that returns one of the five infrastructure keys", () => { const substituting = createGateway({ databaseUrl: "postgres://nobody@example.invalid/none", runtime, @@ -717,12 +718,13 @@ describe("the infrastructure on its own", () => { "db", "agentServer", "publicServer", + "usage", "notes", "worker", ]); }); - it("refuses an extend that returns any of the other three infrastructure keys", () => { + it("refuses an extend that returns any of the other four infrastructure keys", () => { createGateway({ databaseUrl: "postgres://nobody@example.invalid/none", runtime, @@ -750,6 +752,15 @@ describe("the infrastructure on its own", () => { extend: (infra) => ({ publicServer: infra.publicServer }), handlers: () => ({}), }); + createGateway({ + databaseUrl: "postgres://nobody@example.invalid/none", + runtime, + agentListen: { port: 0 }, + publicListen: { port: 0 }, + // @ts-expect-error usage is an infrastructure key and may not be replaced + extend: (infra) => ({ usage: infra.usage }), + handlers: () => ({}), + }); }); }); diff --git a/src/gateway/gateway.ts b/src/gateway/gateway.ts index d5bbe89..48ba4a2 100644 --- a/src/gateway/gateway.ts +++ b/src/gateway/gateway.ts @@ -4,8 +4,14 @@ import type { FastifyInstance, FastifyListenOptions } from "fastify"; import Fastify from "fastify"; import { type Db, openDb } from "../db/index.ts"; import type { Logger } from "../logging/logging.ts"; -import type { Runtime, SignalHandler, SignalHandlers, SignalWorker } from "../signals/index.ts"; -import { createSignalWorker } from "../signals/index.ts"; +import type { + Runtime, + SignalHandler, + SignalHandlers, + SignalWorker, + Usage, +} from "../signals/index.ts"; +import { createSignalWorker, createUsage } from "../signals/index.ts"; import { type Component, createBareGateway, @@ -18,6 +24,7 @@ export type InfraComponents = { db: Db; agentServer: ServerComponent; publicServer: ServerComponent; + usage: Usage; worker: SignalWorker; }; @@ -95,7 +102,9 @@ export function createGateway ...(options.sweepIntervalMs === undefined ? {} : { sweepIntervalMs: options.sweepIntervalMs }), }); - const infra: InfraComponents = { db, agentServer, publicServer, worker }; + const usage = createUsage({ db }); + + const infra: InfraComponents = { db, agentServer, publicServer, usage, worker }; const extension: E = options.extend === undefined ? ({} as E) : options.extend(infra); @@ -103,6 +112,7 @@ export function createGateway db, agentServer, publicServer, + usage, ...extension, worker, } as InfraComponents & E; diff --git a/src/pi/agent-instance.test.ts b/src/pi/agent-instance.test.ts index 0fff56f..b671608 100644 --- a/src/pi/agent-instance.test.ts +++ b/src/pi/agent-instance.test.ts @@ -12,7 +12,7 @@ import { openDb } from "../db/index.ts"; import { createBareGateway, serverComponent } from "../gateway/components.ts"; import type { SignalHandler } from "../signals/handlers.ts"; import * as signalsSchema from "../signals/schema/index.ts"; -import { runs } from "../signals/schema/index.ts"; +import { runs, usage } from "../signals/schema/index.ts"; import { createSignalWorker } from "../signals/worker.ts"; import { applySchema } from "../test-support/apply-schema.ts"; import { createTestDatabase, type TestDatabase } from "../test-support/database.ts"; @@ -36,10 +36,14 @@ import { createPiRuntime } from "./runtime.ts"; const run = promisify(execFile); const skip = await skipContainerTests(); -type Ask = { readonly session: string | null; readonly text: string }; +type Ask = { + readonly session: string | null; + readonly text: string; + readonly usageLimit?: number; +}; const asking: SignalHandler = { - handle: (signal) => [{ session: signal.payload.session, text: signal.payload.text }], + handle: (signal) => [signal.payload], }; const handlerNote = "handler-note.txt"; @@ -73,6 +77,7 @@ type Rig = { runsOf( signalId: string, ): Promise<{ id: string; session: string | null; state: string; error: string | null }[]>; + usageOf(runId?: string): Promise<(typeof usage.$inferSelect)[]>; }; type Paths = { @@ -210,7 +215,7 @@ async function withGateway( const worker = createSignalWorker({ db, runtime, handlers: { ask: asking }, agentServer }); await applySchema(db, signalsSchema); - const handle = db.handle({ runs }); + const handle = db.handle({ runs, usage }); const rig: Rig = { model, agentServerUrl, @@ -239,6 +244,10 @@ async function withGateway( error: row.error, })); }, + async usageOf(runId) { + const all = handle.select().from(usage); + return runId === undefined ? all : all.where(eq(usage.runId, runId)); + }, }; const gateway = createBareGateway({ db, agentServer, worker }); @@ -305,6 +314,18 @@ describe("a real pi behind a real listener", { skip }, () => { ); } + const [firstRun] = await rig.runsOf(first); + assert.ok(firstRun !== undefined); + assert.deepEqual( + (await rig.usageOf(firstRun.id)).map(({ input, output, totalTokens }) => ({ + input, + output, + totalTokens, + })), + Array(3).fill({ input: 1, output: 1, totalTokens: 2 }), + "each of the first Run's three model calls should be recorded with the counts the model sent", + ); + const [freshRun] = await rig.runsOf(fresh); assert.ok(freshRun !== undefined, "the fresh Signal should have one Run"); assert.deepEqual( @@ -358,10 +379,48 @@ describe("a real pi behind a real listener", { skip }, () => { const written = path.join(rig.workspace, agentNote); assert.equal(await readFile(written, "utf8"), "written by the agent"); + const [limited] = await rig.runsOf( + await rig.ask({ session: "user_9", text: "Spend past the limit.", usageLimit: 3 }), + ); + assert.equal(limited?.state, "failed"); + assert.match(limited?.error ?? "", /Usage limit reached: \d+ of 3 tokens/); + assert.ok( + (await rig.usageOf(limited?.id)).length >= 2, + "the calls made before the abort should stay recorded", + ); + const [continued] = await rig.runsOf( + await rig.ask({ session: "user_9", text: "Carry on where you stopped." }), + ); + assert.deepEqual( + [continued?.state, continued?.error], + ["done", null], + "the limited Run's Session should take the next Prompt", + ); + assert.ok( + rig.model.requests.some( + (request) => + request.texts.includes("Carry on where you stopped.") && + request.texts.includes("Spend past the limit."), + ), + "the next Run should carry the limited Run's conversation", + ); + const [failed] = await rig.runsOf(await rig.ask({ session: "user_7", text: doomed })); assert.equal(failed?.state, "failed"); assert.match(failed?.error ?? "", /this deployment has no model/); assert.match(failed?.error ?? "", /stopReason/); + const recorded = await rig.usageOf(); + assert.deepEqual( + recorded.filter((row) => row.totalTokens > 0).map((row) => row.totalTokens), + rig.model.requests + .filter((request) => !request.texts.some((text) => text.includes(doomed))) + .map(() => 2), + "every call the model answered should be one Usage record with the counts it sent", + ); + assert.ok( + recorded.some((row) => row.runId === failed?.id && row.totalTokens === 0), + "the refused call should be recorded too, with nothing charged", + ); const rejectedName = "../escape"; const [rejected] = await rig.runsOf( @@ -372,7 +431,7 @@ describe("a real pi behind a real listener", { skip }, () => { assert.match(rejected?.error ?? "", /Invalid session name/); assert.deepEqual( await rig.sessions(), - [`run_${freshRun.id}.jsonl`, "user_42.jsonl", "user_7.jsonl"].sort(), + [`run_${freshRun.id}.jsonl`, "user_42.jsonl", "user_7.jsonl", "user_9.jsonl"].sort(), "a Session name the framework refused should have left nothing behind", ); }); diff --git a/src/pi/output.test.ts b/src/pi/output.test.ts index 604639a..7334a11 100644 --- a/src/pi/output.test.ts +++ b/src/pi/output.test.ts @@ -3,7 +3,7 @@ import { readFile } from "node:fs/promises"; import { createInterface } from "node:readline"; import { Readable } from "node:stream"; import { describe, it } from "node:test"; -import type { RunOutcome } from "../signals/runtime.ts"; +import type { RunOutcome, UsageReport } from "../signals/runtime.ts"; import { type Framed, framedRecords } from "./framing.ts"; import { readOutcome } from "./output.ts"; @@ -293,3 +293,277 @@ describe("output that cannot be read", () => { ); }); }); + +type Watched = { + readonly outcome: RunOutcome; + readonly reports: readonly UsageReport[]; + readonly aborts: number; +}; + +async function watched( + text: string, + recording: (report: UsageReport) => void | Promise = () => {}, + usageLimit?: number, +): Promise { + const reports: UsageReport[] = []; + let aborts = 0; + const outcome = await readOutcome(framedRecords(chunks(text, 4096)), session, { + observer: { + async usage(report) { + reports.push(report); + await recording(report); + }, + }, + abort: () => { + aborts += 1; + }, + ...(usageLimit === undefined ? {} : { usageLimit }), + }); + return { outcome, reports, aborts }; +} + +function assistantEnd(usage: Record, stopReason = "toolUse"): unknown { + return { + type: "message_end", + message: { role: "assistant", provider: "mock", model: "mock-1", stopReason, usage }, + }; +} + +const spent = { + input: 100, + output: 20, + cacheRead: 300, + cacheWrite: 5, + totalTokens: 425, + cost: { total: 0.25 }, +}; + +describe("the usage a Run reports", () => { + it("reports each model call once, a failed attempt before a retry included", async () => { + const { outcome, reports } = await watched(await fixture("retry-then-settled")); + + assert.deepEqual(outcome, { ok: true }); + assert.deepEqual( + reports.map((report) => [report.provider, report.model, report.totalTokens]), + [ + ["mock", "mock-1", 0], + ["mock", "mock-1", 18], + ], + "two attempts, each reported once although turn_end and agent_end repeat them", + ); + assert.deepEqual(reports[1], { + provider: "mock", + model: "mock-1", + input: 11, + output: 7, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 18, + cost: 0, + }); + }); + + it("carries every category, the cost, and a total it sums itself", async () => { + const { reports } = await watched( + asStream([assistantEnd({ ...spent, totalTokens: 1 }, "stop"), { type: "agent_settled" }]), + ); + + assert.deepEqual(reports, [ + { + provider: "mock", + model: "mock-1", + input: 100, + output: 20, + cacheRead: 300, + cacheWrite: 5, + totalTokens: 425, + cost: 0.25, + }, + ]); + }); + + it("reports a compaction, naming the model the Run last used", async () => { + const { reports } = await watched( + asStream([ + assistantEnd(spent), + { type: "compaction_start", reason: "threshold" }, + { + type: "compaction_end", + reason: "threshold", + result: { summary: "…", usage: { input: 900, output: 60, cacheRead: 0, cacheWrite: 0 } }, + aborted: false, + willRetry: false, + }, + assistantEnd(spent, "stop"), + { type: "agent_settled" }, + ]), + ); + + assert.equal(reports.length, 3); + assert.deepEqual(reports[1], { + provider: "mock", + model: "mock-1", + input: 900, + output: 60, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 960, + cost: 0, + }); + }); + + it("reports nothing for a user message, or a message with no usage", async () => { + const { reports } = await watched( + asStream([ + { type: "message_end", message: { role: "user", usage: spent } }, + { type: "message_end", message: { role: "assistant", stopReason: "stop" } }, + { type: "agent_settled" }, + ]), + ); + + assert.deepEqual(reports, []); + }); + + it("counts a negative or fractional figure as nothing", async () => { + const { reports } = await watched( + asStream([ + assistantEnd({ input: -5, output: 2.5, cacheRead: "9", cacheWrite: 1 }, "stop"), + { type: "agent_settled" }, + ]), + ); + + assert.equal(reports[0]?.totalTokens, 1); + }); +}); + +describe("a usage record that cannot be written", () => { + it("aborts once, reports nothing more, reads to the settle and fails the Run", async () => { + const records = [ + assistantEnd(spent), + assistantEnd(spent), + assistantEnd(spent, "aborted"), + { type: "agent_settled" }, + ]; + let pulledPastSettle = false; + const framed = framedRecords(chunks(asStream([...records, { type: "after" }]), 4096)); + let seen = 0; + const reports: UsageReport[] = []; + let aborts = 0; + + const outcome = await readOutcome( + { + async next() { + seen += 1; + if (seen > records.length) pulledPastSettle = true; + return framed.next(); + }, + }, + session, + { + observer: { + usage(report) { + reports.push(report); + throw new Error("the usage table is gone"); + }, + }, + abort: () => { + aborts += 1; + }, + }, + ); + + assert.equal(aborts, 1); + assert.equal(reports.length, 1); + assert.equal(pulledPastSettle, false); + assert.deepEqual(outcome, { + ok: false, + error: `Session ${session}: Usage recording failed: the usage table is gone`, + }); + }); + + it("fails the Run even when the agent then settled with an answer", async () => { + const { outcome } = await watched(await fixture("settled-ok"), () => { + throw new Error("no database"); + }); + + assert.match(failure(outcome), /Usage recording failed: no database/); + }); +}); + +function failure(outcome: RunOutcome): string { + assert.equal(outcome.ok, false, `this Run should have failed; it was ${JSON.stringify(outcome)}`); + return outcome.ok ? "" : outcome.error; +} + +describe("a Run with a usage limit", () => { + const threeCalls = asStream([ + assistantEnd(spent), + assistantEnd(spent), + assistantEnd(spent, "aborted"), + { type: "agent_settled" }, + ]); + + it("aborts once the call that crossed the limit is recorded, and fails as limited", async () => { + const { outcome, reports, aborts } = await watched(threeCalls, () => {}, 800); + + assert.equal(aborts, 1); + assert.deepEqual(outcome, { + ok: false, + error: `Session ${session}: Usage limit reached: 850 of 800 tokens`, + limited: true, + }); + assert.equal(reports.length, 3, "a call that finished after the abort was still charged"); + }); + + it("lets a Run that reaches the limit exactly, or stays under it, finish as before", async () => { + const stream = asStream([ + assistantEnd(spent), + assistantEnd(spent, "stop"), + { type: "agent_settled" }, + ]); + for (const usageLimit of [850, 1_000_000]) { + const { outcome, aborts } = await watched(stream, () => {}, usageLimit); + assert.deepEqual(outcome, { ok: true }); + assert.equal(aborts, 0); + } + }); + + it("adds a failed attempt and its retry together", async () => { + const { outcome, aborts } = await watched(await fixture("retry-then-settled"), () => {}, 17); + + assert.equal(aborts, 1); + assert.match(failure(outcome), /Usage limit reached: 18 of 17 tokens/); + }); + + it("counts compaction toward the limit", async () => { + const { outcome } = await watched( + asStream([ + assistantEnd(spent), + { + type: "compaction_end", + result: { usage: { input: 500, output: 0, cacheRead: 0, cacheWrite: 0 } }, + }, + assistantEnd(spent, "stop"), + { type: "agent_settled" }, + ]), + () => {}, + 900, + ); + + assert.match(failure(outcome), /925 of 900/); + }); + + it("names the recording failure rather than the limit when recording failed first", async () => { + const { outcome, aborts } = await watched( + threeCalls, + () => { + throw new Error("gone"); + }, + 100, + ); + + assert.equal(aborts, 1); + assert.equal(outcome.ok === false && outcome.limited, undefined); + assert.match(failure(outcome), /Usage recording failed: gone/); + }); +}); diff --git a/src/pi/output.ts b/src/pi/output.ts index bac7659..e04d923 100644 --- a/src/pi/output.ts +++ b/src/pi/output.ts @@ -1,4 +1,4 @@ -import type { RunOutcome } from "../signals/runtime.ts"; +import type { RunObserver, RunOutcome, UsageReport } from "../signals/runtime.ts"; import type { Framed } from "./framing.ts"; const answeredStopReasons = new Set(["stop", "length"]); @@ -8,13 +8,48 @@ type Answer = { readonly errorMessage: string | undefined; }; +export type Watch = { + readonly observer?: RunObserver; + /** Ask the agent to stop, then keep reading until `agent_settled`. */ + readonly abort: () => void; + /** Stop the Run and mark it as failed and limited if total tokens exceed this value. */ + readonly usageLimit?: number; +}; + export async function readOutcome( records: AsyncIterator, session: string, + watch?: Watch, ): Promise { const failed = (why: string): RunOutcome => ({ ok: false, error: `Session ${session}: ${why}` }); let read = 0; let answer: Answer | undefined; + let named: Pick = { provider: null, model: null }; + let stopped: string | undefined; + let limited = false; + let recordingFailed = false; + let used = 0; + + // Report usage even after requesting a stop, since the provider still charges for those calls. + async function report(usage: UsageReport | undefined): Promise { + if (usage === undefined || watch === undefined) return; + used += usage.totalTokens; + if (watch.observer !== undefined && !recordingFailed) { + try { + await watch.observer.usage(usage); + } catch (error) { + recordingFailed = true; + stopped ??= `Usage recording failed: ${error instanceof Error ? error.message : String(error)}`; + watch.abort(); + return; + } + } + if (stopped === undefined && watch.usageLimit !== undefined && used > watch.usageLimit) { + stopped = `Usage limit reached: ${used} of ${watch.usageLimit} tokens`; + limited = true; + watch.abort(); + } + } for (;;) { const step = await records.next(); @@ -36,10 +71,21 @@ export async function readOutcome( read += 1; const record = framed.record; switch (record.type) { - case "message_end": + // Count each call once at message_end; turn_end and agent_end repeat the same message. + case "message_end": { + answer = answerIn(record.message) ?? answer; + const usage = usageIn(record.message, named); + if (usage !== undefined) named = { provider: usage.provider, model: usage.model }; + await report(usage); + break; + } case "turn_end": answer = answerIn(record.message) ?? answer; break; + // Compaction uses a separate model call that is not included in message events. + case "compaction_end": + await report(usageIn(fieldsOf(record.result), named, true)); + break; case "agent_end": if (Array.isArray(record.messages)) { const found = record.messages.map(answerIn).findLast((one) => one !== undefined); @@ -48,7 +94,10 @@ export async function readOutcome( break; // agent_end can precede retries or compaction; only agent_settled completes the run. case "agent_settled": - return settlement(failed, answer, read); + if (stopped === undefined) return settlement(failed, answer, read); + return limited + ? { ok: false, error: `Session ${session}: ${stopped}`, limited } + : failed(stopped); default: break; } @@ -81,6 +130,45 @@ function answerIn(message: unknown): Answer | undefined { }; } +function usageIn( + message: unknown, + named: Pick, + anyRole = false, +): UsageReport | undefined { + const fields = fieldsOf(message); + if (fields === undefined || (!anyRole && fields.role !== "assistant")) return undefined; + const usage = fieldsOf(fields.usage); + if (usage === undefined) return undefined; + const input = count(usage.input); + const output = count(usage.output); + const cacheRead = count(usage.cacheRead); + const cacheWrite = count(usage.cacheWrite); + return { + provider: typeof fields.provider === "string" ? fields.provider : named.provider, + model: typeof fields.model === "string" ? fields.model : named.model, + input, + output, + cacheRead, + cacheWrite, + totalTokens: input + output + cacheRead + cacheWrite, + cost: amount(fieldsOf(usage.cost)?.total), + }; +} + +function fieldsOf(value: unknown): Record | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +function count(value: unknown): number { + return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : 0; +} + +function amount(value: unknown): number { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : 0; +} + function excerpt(line: string): string { const trimmed = line.trim(); return JSON.stringify(trimmed.length > 200 ? `${trimmed.slice(0, 200)}…` : trimmed); diff --git a/src/pi/rpc.ts b/src/pi/rpc.ts index 6368b04..2564a2f 100644 --- a/src/pi/rpc.ts +++ b/src/pi/rpc.ts @@ -10,6 +10,8 @@ export type Answered = { export type RpcChannel = { send(command: string, fields?: Readonly>): Promise; + /** Send a command without waiting for its response. Read the response from `records`. */ + notify(command: string): void; readonly records: AsyncIterator; dropped(): string | undefined; close(): void; @@ -53,6 +55,11 @@ export async function openRpcChannel(host: string, port: number): Promise dropped, close: () => socket.destroy(), + notify(command) { + sent += 1; + socket.write(`${JSON.stringify({ id: String(sent), type: command })}\n`); + }, + async send(command, fields) { sent += 1; const id = String(sent); diff --git a/src/pi/runtime.test.ts b/src/pi/runtime.test.ts index 42e317f..3dbdd5b 100644 --- a/src/pi/runtime.test.ts +++ b/src/pi/runtime.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { createServer } from "node:net"; import { after, describe, it } from "node:test"; -import type { RunOutcome, RunPrompt } from "../signals/runtime.ts"; +import type { RunObserver, RunOutcome, RunPrompt, UsageReport } from "../signals/runtime.ts"; import { dropsTheConnection, type FakeInstance, @@ -33,6 +33,7 @@ async function instanceOf(reply: Reply): Promise { async function runAgainst( reply: Reply, given: RunPrompt = prompt, + observer?: RunObserver, ): Promise<{ outcome: RunOutcome; instance: FakeInstance }> { const instance = await instanceOf(reply); const runtime = createPiRuntime({ @@ -41,7 +42,7 @@ async function runAgainst( sessionsDir, logger: silent, }); - return { outcome: await runtime.run(given), instance }; + return { outcome: await runtime.run(given, observer), instance }; } function failure(outcome: RunOutcome): string { @@ -310,6 +311,100 @@ describe("the sessionsDir an Operator declares", () => { }); }); +describe("the usage a Run reports", () => { + const spending = (stopReason: string): Written => ({ + type: "message_end", + message: { + role: "assistant", + provider: "mock", + model: "mock-1", + stopReason, + usage: { input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }, + }, + }); + + it("hands the observer each model call, and succeeds as before", async () => { + const reports: UsageReport[] = []; + const { outcome } = await runAgainst( + scriptedInstance([spending("toolUse"), spending("stop"), { type: "agent_settled" }]), + prompt, + { usage: (report) => void reports.push(report) }, + ); + + assert.deepEqual(outcome, { ok: true }); + assert.deepEqual( + reports.map((report) => report.totalTokens), + [15, 15], + ); + }); + + it("sends abort when a report cannot be recorded, and fails once the agent settles", { + timeout: 5_000, + }, async () => { + const healthy = scriptedInstance([{ type: "agent_start" }, spending("toolUse")]); + const reply: Reply = (command) => + command.type === "abort" + ? [ + spending("aborted"), + { type: "agent_settled" }, + { type: "response", id: command.id, command: "abort", success: true }, + ] + : healthy(command); + + const { outcome, instance } = await runAgainst(reply, prompt, { + usage: () => { + throw new Error("the Db is down"); + }, + }); + + assert.deepEqual(commandTypes(instance), ["switch_session", "get_state", "prompt", "abort"]); + assert.equal(instance.received[3]?.id, "4"); + assert.equal(failure(outcome), "Session user_42: Usage recording failed: the Db is down"); + }); +}); + +describe("a usage limit", () => { + const spending: Written = { + type: "message_end", + message: { + role: "assistant", + stopReason: "toolUse", + usage: { input: 60, output: 0, cacheRead: 0, cacheWrite: 0 }, + }, + }; + + // The fake agent waits for abort; the test timeout catches a missing abort. + it("sends abort when usage passes the limit, and fails as limited once the agent settles", { + timeout: 5_000, + }, async () => { + const healthy = scriptedInstance([{ type: "agent_start" }, spending, spending]); + const reply: Reply = (command) => + command.type === "abort" + ? [ + { type: "agent_settled" }, + { type: "response", id: command.id, command: "abort", success: true }, + ] + : healthy(command); + + const { outcome, instance } = await runAgainst(reply, { ...prompt, usageLimit: 100 }); + + assert.deepEqual(commandTypes(instance), ["switch_session", "get_state", "prompt", "abort"]); + assert.deepEqual(outcome, { + ok: false, + error: "Session user_42: Usage limit reached: 120 of 100 tokens", + limited: true, + }); + }); + + it("refuses a limit that is not a positive whole number, before connecting", async () => { + for (const usageLimit of [0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY]) { + const { outcome, instance } = await runAgainst(scriptedInstance(), { ...prompt, usageLimit }); + assert.match(failure(outcome), /usageLimit must be a positive whole number/); + assert.equal(instance.connections(), 0); + } + }); +}); + function refusing(command: string, why: string): Reply { const healthy = scriptedInstance(); return (received) => diff --git a/src/pi/runtime.ts b/src/pi/runtime.ts index 78f3fea..0027c40 100644 --- a/src/pi/runtime.ts +++ b/src/pi/runtime.ts @@ -1,6 +1,6 @@ import { posix } from "node:path"; import { defaultLogger, type Logger } from "../logging/index.ts"; -import type { RunOutcome, RunPrompt, Runtime } from "../signals/runtime.ts"; +import type { RunObserver, RunOutcome, RunPrompt, Runtime } from "../signals/runtime.ts"; import { readOutcome } from "./output.ts"; import { openRpcChannel, type RpcChannel } from "./rpc.ts"; @@ -22,6 +22,12 @@ const sessionNames = /^[A-Za-z0-9](?:[A-Za-z0-9._-]*[A-Za-z0-9])?$/; * session names must begin and end with an ASCII letter or digit; interior characters may also * include `.`, `_`, and `-`. Blank prompt text fails the run. Completion requires `agent_settled` * and a final assistant stop reason of `stop` or `length`. There is no execution timeout. + * + * Reports usage for each assistant `message_end` and each `compaction_end`, including retries. + * If the observer throws or usage exceeds the Prompt's `usageLimit`, the runtime sends `abort` + * to pi. Once pi settles, the Run fails and the Session file is kept. Token counts come from + * the provider. A response interrupted mid-stream reports less output than was generated. + * If the server reports no usage, the recorded count is zero. */ export function createPiRuntime(instance: PiInstance): Runtime { const { host, port, sessionsDir } = instance; @@ -39,7 +45,7 @@ export function createPiRuntime(instance: PiInstance): Runtime { } return { - async run(prompt: RunPrompt): Promise { + async run(prompt: RunPrompt, observer?: RunObserver): Promise { const failed = (why: string): RunOutcome => ({ ok: false, error: `Session ${prompt.session}: ${why}`, @@ -53,6 +59,14 @@ export function createPiRuntime(instance: PiInstance): Runtime { if (prompt.text.trim() === "") { return failed("Prompt text must not be blank"); } + if ( + prompt.usageLimit !== undefined && + !(Number.isSafeInteger(prompt.usageLimit) && prompt.usageLimit > 0) + ) { + return failed( + `usageLimit must be a positive whole number of tokens; received ${String(prompt.usageLimit)}`, + ); + } const sessionFile = posix.join(sessionsDir, `${prompt.session}.jsonl`); @@ -66,7 +80,7 @@ export function createPiRuntime(instance: PiInstance): Runtime { log.debug({ session: prompt.session, sessionFile, host, port }, "connected to the agent"); try { - return await performRun(rpc, prompt, sessionFile, failed); + return await performRun(rpc, prompt, sessionFile, failed, observer); } catch (error) { return failed(messageOf(error)); } finally { @@ -82,6 +96,7 @@ async function performRun( prompt: RunPrompt, sessionFile: string, failed: (why: string) => RunOutcome, + observer: RunObserver | undefined, ): Promise { const switched = await rpc.send("switch_session", { sessionPath: sessionFile }); if (!switched.success) { @@ -110,7 +125,17 @@ async function performRun( return failed(`Prompt rejected: ${prompted.error ?? "no error details"}`); } - const outcome = await readOutcome(rpc.records, prompt.session); + let aborted = false; + const abort = (): void => { + if (aborted) return; + aborted = true; + rpc.notify("abort"); + }; + const outcome = await readOutcome(rpc.records, prompt.session, { + ...(observer === undefined ? {} : { observer }), + ...(prompt.usageLimit === undefined ? {} : { usageLimit: prompt.usageLimit }), + abort, + }); const dropped = rpc.dropped(); if (outcome.ok || dropped === undefined) return outcome; return { ok: false, error: `${outcome.error}. The connection failed: ${dropped}` }; diff --git a/src/signals/handlers.ts b/src/signals/handlers.ts index a649fba..93f7209 100644 --- a/src/signals/handlers.ts +++ b/src/signals/handlers.ts @@ -17,10 +17,27 @@ export type Prompt = { * Omission becomes `null`. The runtime does not receive it; put agent instructions in `text`. */ readonly context?: unknown; + /** + * The User whose usage totals include this Run. Use data authenticated by the Gateway, such + * as a Message's `userId`. Never use text written by a model or sender. The agent does not + * receive this field. It is separate from the Session name and prompt text. If omitted, + * the Run's usage is not assigned to any User. + */ + readonly userId?: string; + /** + * The maximum `totalTokens` for this Run, including compaction. If the total exceeds this + * limit, the runtime stops the agent and marks the Run as failed and limited. The Session + * keeps its history up to that point. Usage is reported after each model call finishes, so + * the Run can exceed the limit by at most one call. Must be a positive whole number. Omit + * this field for no limit. To enforce a budget, pass the remaining token allowance. + */ + readonly usageLimit?: number; }; export type PostOutcome = { readonly failed: boolean; + /** True if any Run stopped because it exceeded its `usageLimit`. That Run also failed. */ + readonly limited: boolean; }; /** @@ -28,8 +45,8 @@ export type PostOutcome = { * a failed run does not prevent later prompts from running. * * `post` runs after all runs, including when `handle` throws. `failed` covers failures in - * `handle` or any run. A thrown `post` also fails the signal. No phase is automatically retried, - * and the worker imposes no timeout. + * `handle` or any run. `limited` is true if any run exceeded its usage limit. If `post` throws, + * the signal also fails. No phase is automatically retried, and the worker imposes no timeout. */ export type SignalHandler = { handle(signal: Signal): readonly Prompt[] | Promise; diff --git a/src/signals/index.ts b/src/signals/index.ts index 90a3615..fa23e8a 100644 --- a/src/signals/index.ts +++ b/src/signals/index.ts @@ -6,11 +6,13 @@ export type { SignalHandlers, } from "./handlers.ts"; export type { RunRecord, SignalRecord } from "./routes.ts"; -export type { RunOutcome, RunPrompt, Runtime } from "./runtime.ts"; +export type { RunObserver, RunOutcome, RunPrompt, Runtime, UsageReport } from "./runtime.ts"; export type { RunState, SignalState } from "./schema/index.ts"; export { runStates, signalStates } from "./schema/index.ts"; export type { TemplateHandlerOptions } from "./template-handler.ts"; export { templateHandler } from "./template-handler.ts"; +export type { Usage, UsageOptions, UsageQuery, UsageTotals } from "./usage.ts"; +export { createUsage } from "./usage.ts"; export type { CurrentRun, EmittedSignal, diff --git a/src/signals/runtime.ts b/src/signals/runtime.ts index 86cccc4..8f2e2a0 100644 --- a/src/signals/runtime.ts +++ b/src/signals/runtime.ts @@ -1,16 +1,51 @@ import type { Prompt } from "./handlers.ts"; -export type RunOutcome = { readonly ok: true } | { readonly ok: false; readonly error: string }; +export type RunOutcome = + | { readonly ok: true } + | { + readonly ok: false; + readonly error: string; + /** The Run was stopped because it exceeded its `usageLimit`. */ + readonly limited?: boolean; + }; -export type RunPrompt = Omit & { +export type RunPrompt = Omit & { readonly session: string; }; +/** Usage for one model call, reported by the provider to the agent implementation. */ +export type UsageReport = { + /** `null` when the provider is unknown, such as for some summarization calls. */ + readonly provider: string | null; + readonly model: string | null; + /** Prompt tokens not served from the provider's cache. */ + readonly input: number; + readonly output: number; + readonly cacheRead: number; + readonly cacheWrite: number; + /** `input + output + cacheRead + cacheWrite`. */ + readonly totalTokens: number; + /** Cost calculated by the agent implementation; 0 if the model price is unknown. */ + readonly cost: number; +}; + +/** Supplied by the worker for one Run. */ +export type RunObserver = { + /** + * The runtime calls and awaits this method once per model call, in order. If it throws, + * the runtime must stop and fail the Run to prevent further unrecorded usage. + */ + usage(report: UsageReport): void | Promise; +}; + /** * The worker calls `run` serially. A thrown error is recorded as a failed run. * The framework supplies no timeout or cancellation; an unsettled call blocks subsequent signals. * Agent output is not delivered to users automatically. Use the messenger to send messages. + * + * The worker always passes an `observer`. If a runtime never reports usage, no usage or cost + * is recorded for its Runs. */ export type Runtime = { - run(prompt: RunPrompt): Promise; + run(prompt: RunPrompt, observer?: RunObserver): Promise; }; diff --git a/src/signals/schema/index.ts b/src/signals/schema/index.ts index 69f4f8a..5cd4b55 100644 --- a/src/signals/schema/index.ts +++ b/src/signals/schema/index.ts @@ -2,7 +2,9 @@ import { type SQL, sql } from "drizzle-orm"; import { check, index, + integer, jsonb, + numeric, type PgColumn, pgSchema, text, @@ -51,6 +53,8 @@ export const runs = signalsSchema.table( session: text("session"), prompt: text("prompt").notNull(), context: jsonb("context"), + /** Not a foreign key: Signals does not depend on Users. */ + userId: text("user_id"), state: text("state").$type().notNull().default("pending"), error: text("error"), startedAt: timestamp("started_at", { withTimezone: true }), @@ -62,4 +66,32 @@ export const runs = signalsSchema.table( ], ); -export const signalsTables = { signals, runs }; +/** One row per model call, saved when usage is reported and never updated. */ +export const usage = signalsSchema.table( + "usage", + { + id: uuid("id").primaryKey().defaultRandom(), + runId: uuid("run_id") + .notNull() + .references(() => runs.id), + /** Copy the Run's `userId` so User totals can be calculated without joining the runs table. */ + userId: text("user_id"), + provider: text("provider"), + model: text("model"), + input: integer("input").notNull(), + output: integer("output").notNull(), + cacheRead: integer("cache_read").notNull(), + cacheWrite: integer("cache_write").notNull(), + totalTokens: integer("total_tokens").notNull(), + cost: numeric("cost", { mode: "number" }).notNull(), + recordedAt: timestamp("recorded_at", { withTimezone: true }) + .notNull() + .default(sql`clock_timestamp()`), + }, + (table) => [ + index("usage_run_idx").on(table.runId), + index("usage_user_idx").on(table.userId, table.recordedAt), + ], +); + +export const signalsTables = { signals, runs, usage }; diff --git a/src/signals/template-handler.test.ts b/src/signals/template-handler.test.ts index 5c0ab0c..0608160 100644 --- a/src/signals/template-handler.test.ts +++ b/src/signals/template-handler.test.ts @@ -124,6 +124,49 @@ describe("the template Handler", () => { }); }); +describe("the User a Run is charged to", () => { + type Sent = { readonly userId: string }; + + it("writes what the user function answers onto the Prompt, awaited if it has to be", async () => { + for (const user of [ + (signal: Signal) => signal.payload.userId, + async (signal: Signal) => signal.payload.userId, + ]) { + const handler = templateHandler({ + template: "for {{userId}}", + session: (signal) => `user_${signal.payload.userId}`, + data: (signal) => signal.payload, + user, + }); + + assert.deepEqual(await onlyPrompt(handler, { userId: "42" }), { + session: "user_42", + text: "for 42", + userId: "42", + }); + } + }); + + it("charges no User when the function answers null or is not given", async () => { + const answeringNull = templateHandler({ + template: "x", + session: () => null, + data: () => ({}), + user: () => null, + }); + const notGiven = templateHandler({ + template: "x", + session: () => null, + data: () => ({}), + }); + + for (const handler of [answeringNull, notGiven]) { + const prompt = await onlyPrompt(handler, { userId: "42" }); + assert.equal(Object.hasOwn(prompt, "userId"), false); + } + }); +}); + describe("a substituted value", () => { it("reaches the Prompt byte for byte, escaping nothing", async () => { const text = await renders("{{value}}", { value: everythingHandlebarsWouldEscape }); @@ -328,8 +371,8 @@ describe("the template Handler's post phase", () => { assert.equal(handler.post, post, "the function should reach the Handler as it was written"); const signal = aSignal({ userId: "u_1" }); - await handler.post?.(signal, { failed: true }); - assert.deepEqual(seen, [{ signal, outcome: { failed: true } }]); + await handler.post?.(signal, { failed: true, limited: false }); + assert.deepEqual(seen, [{ signal, outcome: { failed: true, limited: false } }]); }); }); @@ -461,8 +504,8 @@ describe("the template Handler under the worker", () => { assert.equal((await settled(unsupplied)).state, "failed"); assert.deepEqual(outcomes, { - "prompt.good": { failed: false }, - "prompt.unsupplied": { failed: true }, + "prompt.good": { failed: false, limited: false }, + "prompt.unsupplied": { failed: true, limited: false }, }); assert.deepEqual(runtime.texts(), ["a Prompt that renders"]); }); diff --git a/src/signals/template-handler.ts b/src/signals/template-handler.ts index 1c65107..dc67a72 100644 --- a/src/signals/template-handler.ts +++ b/src/signals/template-handler.ts @@ -11,6 +11,9 @@ export type TemplateHandlerOptions = { readonly context?: (signal: Signal) => unknown; + /** Sets `Prompt.userId`. Return `null` or omit this function to assign no User to the Run. */ + readonly user?: (signal: Signal) => string | null | Promise; + readonly helpers?: Readonly>; readonly partials?: Readonly>; @@ -46,6 +49,8 @@ export function templateHandler( const data = await options.data(signal); const context = options.context === undefined ? {} : { context: await options.context(signal) }; + const userId = options.user === undefined ? null : await options.user(signal); + const charged = userId === null ? {} : { userId }; let text: string; try { @@ -56,7 +61,7 @@ export function templateHandler( }); } - return [{ session, text, ...context }]; + return [{ session, text, ...context, ...charged }]; }, ...(options.post === undefined ? {} : { post: options.post }), diff --git a/src/signals/usage.test.ts b/src/signals/usage.test.ts new file mode 100644 index 0000000..ff7d102 --- /dev/null +++ b/src/signals/usage.test.ts @@ -0,0 +1,183 @@ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { after, before, describe, it } from "node:test"; +import { type Db, openDb } from "../db/index.ts"; +import { applySchema } from "../test-support/apply-schema.ts"; +import { createTestDatabase, type TestDatabase } from "../test-support/database.ts"; +import * as signalsSchema from "./schema/index.ts"; +import { runs, signals, signalsTables, usage } from "./schema/index.ts"; +import { createUsage } from "./usage.ts"; + +let database: TestDatabase; +let db: Db; + +const monday = new Date("2026-10-26T00:00:00Z"); +const tuesday = new Date("2026-10-27T00:00:00Z"); +const wednesday = new Date("2026-10-28T00:00:00Z"); + +before(async () => { + database = await createTestDatabase("usage"); + db = database.db; + await applySchema(db, signalsSchema); + + const handle = db.handle(signalsTables); + const [signal] = await handle + .insert(signals) + .values({ kind: "spent", payload: {} }) + .returning({ id: signals.id }); + assert.ok(signal); + const runId = randomUUID(); + await handle.insert(runs).values({ id: runId, signalId: signal.id, prompt: "spend" }); + + const call = (recordedAt: Date, input: number, output: number, cost: number) => ({ + runId, + userId: null, + provider: "mock", + model: "mock-1", + input, + output, + cacheRead: 1000, + cacheWrite: 10, + totalTokens: input + output + 1010, + cost, + recordedAt, + }); + await handle + .insert(usage) + .values([ + call(new Date("2026-10-26T09:00:00Z"), 100, 20, 0.125), + call(new Date("2026-10-26T23:59:59Z"), 50, 5, 0.25), + call(new Date("2026-10-27T00:00:00Z"), 7, 3, 0.5), + ]); + + const alicesRun = randomUUID(); + const bobsRun = randomUUID(); + await handle.insert(runs).values([ + { id: alicesRun, signalId: signal.id, prompt: "alice", userId: "alice" }, + { id: bobsRun, signalId: signal.id, prompt: "bob", userId: "bob" }, + ]); + const charged = (id: string, userId: string, recordedAt: Date, totalTokens: number) => ({ + runId: id, + userId, + provider: "mock", + model: "mock-1", + input: totalTokens, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens, + cost: 0, + recordedAt, + }); + await handle + .insert(usage) + .values([ + charged(alicesRun, "alice", new Date("2026-10-26T10:00:00Z"), 40_000), + charged(alicesRun, "alice", new Date("2026-10-27T10:00:00Z"), 15_000), + charged(alicesRun, "alice", new Date("2026-10-27T11:00:00Z"), 6_000), + charged(bobsRun, "bob", new Date("2026-10-27T12:00:00Z"), 900), + ]); +}); + +after(() => database.drop()); + +describe("usage.total", () => { + it("sums every category, the cost and the calls across all records", async () => { + assert.deepEqual(await createUsage({ db }).total({ userId: null }), { + input: 157, + output: 28, + cacheRead: 3000, + cacheWrite: 30, + totalTokens: 3215, + cost: 0.875, + calls: 3, + }); + }); + + it("counts since inclusively and until exclusively, which is a day window", async () => { + const onMonday = await createUsage({ db }).total({ + userId: null, + since: monday, + until: tuesday, + }); + const onTuesday = await createUsage({ db }).total({ + userId: null, + since: tuesday, + until: wednesday, + }); + + assert.deepEqual([onMonday.calls, onMonday.totalTokens], [2, 2195]); + assert.deepEqual([onTuesday.calls, onTuesday.totalTokens], [1, 1020]); + }); + + it("starts from nothing again once the window moves past the records, which is a reset", async () => { + assert.deepEqual(await createUsage({ db }).total({ since: wednesday }), { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: 0, + calls: 0, + }); + }); + + it("reads the same totals from a fresh component on the same Db, which is a restart", async () => { + const before = await createUsage({ db }).total(); + const again = createUsage({ db }); + await again.start(); + + assert.deepEqual(await again.total(), before); + await again.stop(); + }); +}); + +describe("usage.total for one User", () => { + const limit = 20_000; + + it("counts that User's records only, so one User spending leaves another untouched", async () => { + const alice = await createUsage({ db }).total({ userId: "alice", since: tuesday }); + const bob = await createUsage({ db }).total({ userId: "bob", since: tuesday }); + + assert.deepEqual([alice.totalTokens, alice.calls], [21_000, 2]); + assert.deepEqual([bob.totalTokens, bob.calls], [900, 1]); + assert.ok(alice.totalTokens >= limit && bob.totalTokens < limit); + }); + + it("counts a User with no records as zero", async () => { + assert.equal((await createUsage({ db }).total({ userId: "carol" })).calls, 0); + }); + + it("counts Runs charged to no User under null and everything when userId is omitted", async () => { + const reader = createUsage({ db }); + const [nobody, alice, bob, everyone] = await Promise.all([ + reader.total({ userId: null }), + reader.total({ userId: "alice" }), + reader.total({ userId: "bob" }), + reader.total(), + ]); + + assert.equal(everyone.totalTokens, nobody.totalTokens + alice.totalTokens + bob.totalTokens); + }); + + it("starts the User from zero in a new window, which is the daily reset", async () => { + const yesterday = await createUsage({ db }).total({ + userId: "alice", + since: monday, + until: tuesday, + }); + + assert.equal(yesterday.totalTokens, 40_000); + assert.equal((await createUsage({ db }).total({ userId: "alice", since: wednesday })).calls, 0); + }); + + it("reads the same per-User totals over a new connection pool, which is a restart", async () => { + const before = await createUsage({ db }).total({ userId: "alice" }); + const reopened = openDb(database.url); + try { + assert.deepEqual(await createUsage({ db: reopened }).total({ userId: "alice" }), before); + } finally { + await reopened.stop(); + } + }); +}); diff --git a/src/signals/usage.ts b/src/signals/usage.ts new file mode 100644 index 0000000..231f0be --- /dev/null +++ b/src/signals/usage.ts @@ -0,0 +1,97 @@ +import { type AnyColumn, and, eq, gte, isNull, lt, type SQL, sql } from "drizzle-orm"; +import type { Db } from "../db/index.ts"; +import type { Component } from "../gateway/components.ts"; +import type { WorkerHandle } from "./routes.ts"; +import type { UsageReport } from "./runtime.ts"; +import { signalsTables, usage } from "./schema/index.ts"; + +export type UsageOptions = { + readonly db: Db; +}; + +export type UsageQuery = { + /** Filter by User. Use `null` for Runs with no User, or omit to include all Users and Runs. */ + readonly userId?: string | null; + /** Include records at or after this time. Omit to start from the first record. */ + readonly since?: Date; + /** Exclusive. Omitted means up to now. */ + readonly until?: Date; +}; + +export type UsageTotals = { + readonly input: number; + readonly output: number; + readonly cacheRead: number; + readonly cacheWrite: number; + readonly totalTokens: number; + readonly cost: number; + /** Number of model calls included in the totals. */ + readonly calls: number; +}; + +/** + * Reads the Usage records saved by the Signal Worker, one per model call. Saved records + * survive restarts. Token counts come from the provider and do not directly measure cost. + * The accuracy of `cost` depends on the model price used by the agent implementation. + * `createGateway` constructs this component for you. + */ +export type Usage = Component & { + /** Calculate totals in the database for records matching the User and time filters. */ + total(query?: UsageQuery): Promise; +}; + +export function createUsage(options: UsageOptions): Usage { + const handle = options.db.handle(signalsTables); + + return { + async start() {}, + async stop() {}, + + async total(query = {}) { + const bounds: SQL[] = []; + if (query.userId === null) bounds.push(isNull(usage.userId)); + else if (query.userId !== undefined) bounds.push(eq(usage.userId, query.userId)); + if (query.since !== undefined) bounds.push(gte(usage.recordedAt, query.since)); + if (query.until !== undefined) bounds.push(lt(usage.recordedAt, query.until)); + + const [sums] = await handle + .select({ + input: summed(usage.input), + output: summed(usage.output), + cacheRead: summed(usage.cacheRead), + cacheWrite: summed(usage.cacheWrite), + totalTokens: summed(usage.totalTokens), + cost: summed(usage.cost), + calls: sql`count(*)`.mapWith(Number), + }) + .from(usage) + .where(and(...bounds)); + if (sums === undefined) throw new Error("A usage sum returned no row"); + return sums; + }, + }; +} + +function summed(column: AnyColumn): SQL { + return sql`coalesce(sum(${column}), 0)`.mapWith(Number); +} + +export async function recordUsage( + handle: WorkerHandle, + runId: string, + userId: string | null, + report: UsageReport, +): Promise { + await handle.insert(usage).values({ + runId, + userId, + provider: report.provider, + model: report.model, + input: report.input, + output: report.output, + cacheRead: report.cacheRead, + cacheWrite: report.cacheWrite, + totalTokens: report.totalTokens, + cost: report.cost, + }); +} diff --git a/src/signals/worker.test.ts b/src/signals/worker.test.ts index d19346d..3f93056 100644 --- a/src/signals/worker.test.ts +++ b/src/signals/worker.test.ts @@ -10,9 +10,9 @@ import { createTestDatabase, type TestDatabase } from "../test-support/database. import { fakeRuntime } from "../test-support/fake-runtime.ts"; import { waitUntil } from "../test-support/wait.ts"; import type { SignalHandler, SignalHandlers } from "./handlers.ts"; -import type { Runtime } from "./runtime.ts"; +import type { Runtime, UsageReport } from "./runtime.ts"; import * as signalsSchema from "./schema/index.ts"; -import { runs, signals } from "./schema/index.ts"; +import { runs, signals, usage } from "./schema/index.ts"; import { createSignalWorker, type SignalWorker, signalChannel } from "./worker.ts"; let database: TestDatabase; @@ -623,6 +623,232 @@ describe("the worker", () => { }); }); +describe("usage", () => { + const call = (totalTokens: number): UsageReport => ({ + provider: "mock", + model: "mock-1", + input: totalTokens, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens, + cost: 0.5, + }); + + function usageOf(runId: string): Promise<(typeof usage.$inferSelect)[]> { + return db + .handle({ usage }) + .select() + .from(usage) + .where(eq(usage.runId, runId)) + .orderBy(asc(usage.recordedAt)); + } + + it("records each model call a Run reports against that Run, as it reports it", async () => { + let recordedBeforeReturning = -1; + const runtime = fakeRuntime(async (prompt, observer) => { + const sizes = prompt.text === "first" ? [10, 20] : [5]; + for (const size of sizes) await observer?.usage(call(size)); + const [run] = await db.handle({ runs }).select().from(runs).where(eq(runs.state, "running")); + if (prompt.text === "first" && run !== undefined) { + recordedBeforeReturning = (await usageOf(run.id)).length; + } + return { ok: true }; + }); + await withWorker( + { + spending: { + handle: () => [ + { session: null, text: "first" }, + { session: null, text: "second" }, + ], + }, + }, + runtime, + async (worker) => { + const id = await emit(worker, "spending"); + assert.equal((await settled(id)).state, "done"); + + const [first, second] = await runsOf(id); + assert.ok(first && second); + assert.deepEqual( + (await usageOf(first.id)).map((row) => [ + row.provider, + row.model, + row.totalTokens, + row.cost, + ]), + [ + ["mock", "mock-1", 10, 0.5], + ["mock", "mock-1", 20, 0.5], + ], + ); + assert.deepEqual( + (await usageOf(second.id)).map((row) => row.totalTokens), + [5], + ); + assert.equal(recordedBeforeReturning, 2, "each record is written as reported"); + }, + ); + }); + + it("stores the User a Prompt names on its Run and on every record of it", async () => { + const runtime = fakeRuntime(async (_prompt, observer) => { + await observer?.usage(call(3)); + await observer?.usage(call(4)); + return { ok: true }; + }); + await withWorker( + { + charged: { + handle: () => [ + { session: "user_alice", text: "for alice", userId: "alice" }, + { session: null, text: "for nobody" }, + ], + }, + }, + runtime, + async (worker) => { + const id = await emit(worker, "charged"); + assert.equal((await settled(id)).state, "done"); + + const [alices, nobodys] = await runsOf(id); + assert.ok(alices && nobodys); + assert.deepEqual([alices.userId, nobodys.userId], ["alice", null]); + assert.deepEqual( + (await usageOf(alices.id)).map((row) => row.userId), + ["alice", "alice"], + ); + assert.deepEqual( + (await usageOf(nobodys.id)).map((row) => row.userId), + [null, null], + ); + assert.equal( + Object.hasOwn(runtime.recorded[0] ?? {}, "userId"), + false, + "the Runtime is not told whom the Run is charged to", + ); + }, + ); + }); + + it("hands the Runtime the Prompt's usage limit, and only when there is one", async () => { + const runtime = fakeRuntime(); + await withWorker( + { + capped: { + handle: () => [ + { session: "s", text: "capped", usageLimit: 500 }, + { session: "s", text: "uncapped" }, + ], + }, + }, + runtime, + async (worker) => { + await settled(await emit(worker, "capped")); + assert.deepEqual(runtime.recorded, [ + { session: "s", text: "capped", usageLimit: 500 }, + { session: "s", text: "uncapped" }, + ]); + }, + ); + }); + + it("tells the post phase a Run was limited, and records why it failed", async () => { + const told: { failed: boolean; limited: boolean }[] = []; + const runtime = fakeRuntime((prompt) => + prompt.text === "too much" + ? { ok: false, error: "Usage limit reached: 120 of 100 tokens", limited: true } + : prompt.text === "broken" + ? { ok: false, error: "the model is down" } + : { ok: true }, + ); + const posting = (texts: string[]): SignalHandler => ({ + handle: () => texts.map((text) => ({ session: "s", text })), + post: (_signal, outcome) => { + told.push({ failed: outcome.failed, limited: outcome.limited }); + }, + }); + await withWorker( + { + limited: posting(["fine", "too much"]), + failed: posting(["broken"]), + fine: posting(["fine"]), + }, + runtime, + async (worker) => { + const limitedId = await emit(worker, "limited"); + await settled(limitedId); + await settled(await emit(worker, "failed")); + await settled(await emit(worker, "fine")); + + assert.deepEqual(told, [ + { failed: true, limited: true }, + { failed: true, limited: false }, + { failed: false, limited: false }, + ]); + const [, stopped] = await runsOf(limitedId); + assert.deepEqual( + [stopped?.state, stopped?.error], + ["failed", "Usage limit reached: 120 of 100 tokens"], + ); + }, + ); + }); + + it("keeps the records of a Run that then failed", async () => { + const runtime = fakeRuntime(async (_prompt, observer) => { + await observer?.usage(call(7)); + return { ok: false, error: "the model gave up" }; + }); + await withWorker( + { costly: { handle: () => [{ session: null, text: "spend then fail" }] } }, + runtime, + async (worker) => { + const id = await emit(worker, "costly"); + assert.equal((await settled(id)).state, "failed"); + const [run] = await runsOf(id); + assert.ok(run); + assert.deepEqual( + (await usageOf(run.id)).map((row) => row.totalTokens), + [7], + ); + }, + ); + }); + + it("throws into the Runtime when a record cannot be written, so the Run fails closed", async () => { + let thrown: unknown; + const runtime = fakeRuntime(async (_prompt, observer) => { + try { + await observer?.usage(call(9)); + } catch (error) { + thrown = error; + return { ok: false, error: `Usage recording failed: ${(error as Error).message}` }; + } + return { ok: true }; + }); + const tables = db.handle({}); + await tables.execute(sql`alter table concorde_signals.usage rename to usage_gone`); + try { + await withWorker( + { unrecorded: { handle: () => [{ session: null, text: "spend" }] } }, + runtime, + async (worker) => { + const id = await emit(worker, "unrecorded"); + const row = await settled(id); + + assert.ok(thrown instanceof Error); + assert.equal(row.state, "failed"); + assert.match(row.error ?? "", /Usage recording failed/); + }, + ); + } finally { + await tables.execute(sql`alter table concorde_signals.usage_gone rename to usage`); + } + }); +}); + describe("wakeup", () => { const one: SignalHandlers = { woken: { handle: () => [{ session: null, text: "woken" }] } }; diff --git a/src/signals/worker.ts b/src/signals/worker.ts index 44db76f..73ac2b3 100644 --- a/src/signals/worker.ts +++ b/src/signals/worker.ts @@ -8,6 +8,7 @@ import type { Prompt, Signal, SignalHandlers } from "./handlers.ts"; import { agentReadRoutes } from "./routes.ts"; import type { RunOutcome, RunPrompt, Runtime } from "./runtime.ts"; import { runs, signals, signalsTables } from "./schema/index.ts"; +import { recordUsage } from "./usage.ts"; export type EmittedSignal = { readonly kind: string; @@ -73,6 +74,7 @@ type QueuedRun = { readonly id: string; readonly prompt: RunPrompt; readonly context: unknown; + readonly userId: string | null; }; /** @@ -154,6 +156,7 @@ export function createSignalWorker(options: SignalWorkerOptions): SignalWorker { let prompts: readonly Prompt[] | undefined; let failure: string | undefined; + let limited = false; try { prompts = await handler.handle(signal); } catch (error) { @@ -163,12 +166,12 @@ export function createSignalWorker(options: SignalWorkerOptions): SignalWorker { } if (prompts !== undefined) { - failure = await executeRuns(signal, prompts); + ({ failure, limited } = await executeRuns(signal, prompts)); } if (handler.post !== undefined) { try { - await handler.post(signal, { failed: failure !== undefined }); + await handler.post(signal, { failed: failure !== undefined, limited }); } catch (error) { log.error( { signalId: signal.id, kind: signal.kind, err: error }, @@ -185,40 +188,52 @@ export function createSignalWorker(options: SignalWorkerOptions): SignalWorker { async function executeRuns( signal: Signal, prompts: readonly Prompt[], - ): Promise { + ): Promise<{ failure: string | undefined; limited: boolean }> { if (prompts.length === 0) { - return undefined; + return { failure: undefined, limited: false }; } const queued: QueuedRun[] = prompts.map((prompt) => { const id = randomUUID(); - return { id, prompt: promptForRun(id, prompt), context: prompt.context ?? null }; + return { + id, + prompt: promptForRun(id, prompt), + context: prompt.context ?? null, + userId: prompt.userId ?? null, + }; }); await handle.insert(runs).values( - queued.map(({ id, prompt, context }) => ({ + queued.map(({ id, prompt, context, userId }) => ({ id, signalId: signal.id, session: prompt.session, prompt: prompt.text, context, + userId, })), ); const failures: string[] = []; + let limited = false; for (const run of queued) { const outcome = await executeRun(signal, run); - if (!outcome.ok) failures.push(outcome.error); + if (!outcome.ok) { + failures.push(outcome.error); + limited ||= outcome.limited === true; + } } - if (failures.length === 0) return undefined; + if (failures.length === 0) return { failure: undefined, limited }; const first = failures[0] ?? ""; - return failures.length === 1 - ? `Run failed: ${first}` - : `${failures.length} of ${queued.length} runs failed. First error: ${first}`; + const failure = + failures.length === 1 + ? `Run failed: ${first}` + : `${failures.length} of ${queued.length} runs failed. First error: ${first}`; + return { failure, limited }; } async function executeRun(signal: Signal, run: QueuedRun): Promise { - const { id: runId, prompt, context } = run; + const { id: runId, prompt, context, userId } = run; log.info({ runId, signalId: signal.id, session: prompt.session }, "Run started"); await handle .update(runs) @@ -228,7 +243,9 @@ export function createSignalWorker(options: SignalWorkerOptions): SignalWorker { current = { id: runId, signalId: signal.id, session: prompt.session, context }; let outcome: RunOutcome; try { - outcome = await runtime.run(prompt); + outcome = await runtime.run(prompt, { + usage: (report) => recordUsage(handle, runId, userId, report), + }); } catch (error) { outcome = { ok: false, error: describe(error) }; } finally { @@ -377,7 +394,10 @@ export function createSignalWorker(options: SignalWorkerOptions): SignalWorker { } function promptForRun(runId: string, prompt: Prompt): RunPrompt { - return { session: prompt.session ?? `run_${runId}`, text: prompt.text }; + const session = prompt.session ?? `run_${runId}`; + return prompt.usageLimit === undefined + ? { session, text: prompt.text } + : { session, text: prompt.text, usageLimit: prompt.usageLimit }; } function describe(error: unknown): string { diff --git a/src/test-support/fake-runtime.ts b/src/test-support/fake-runtime.ts index 441a3ea..8e13c89 100644 --- a/src/test-support/fake-runtime.ts +++ b/src/test-support/fake-runtime.ts @@ -1,6 +1,9 @@ -import type { RunOutcome, RunPrompt, Runtime } from "../signals/runtime.ts"; +import type { RunObserver, RunOutcome, RunPrompt, Runtime } from "../signals/runtime.ts"; -export type RuntimeScript = (prompt: RunPrompt) => RunOutcome | Promise; +export type RuntimeScript = ( + prompt: RunPrompt, + observer: RunObserver | undefined, +) => RunOutcome | Promise; export type FakeRuntime = Runtime & { readonly recorded: readonly RunPrompt[]; @@ -23,14 +26,14 @@ export function fakeRuntime(script: RuntimeScript = () => ({ ok: true })): FakeR texts() { return recorded.map((prompt) => prompt.text); }, - async run(prompt) { + async run(prompt, observer) { recorded.push(prompt); inFlight += 1; if (inFlight > 1) overlapped = true; try { await new Promise((resume) => setTimeout(resume, holdMs)); if (inFlight > 1) overlapped = true; - return await script(prompt); + return await script(prompt, observer); } finally { inFlight -= 1; } From be5e66e68b88b10666b72458a6a751b9d1df5ed8 Mon Sep 17 00:00:00 2001 From: blockchainluffy Date: Tue, 27 Oct 2026 14:40:38 +0530 Subject: [PATCH 3/3] chore: simplify usage documentation and comments Co-Authored-By: GPT-6 --- CLAUDE.md | 19 +++++++------- CONTEXT.md | 37 +++++++++++++++++++++++---- README.md | 2 +- examples/05_usage_limits/AGENTS.md | 17 +++++++------ examples/05_usage_limits/README.md | 40 ++++++++++++++++-------------- scripts/check-package.ts | 21 ++++++++-------- site/architecture.md | 4 +-- src/pi/output.ts | 6 ++--- src/pi/runtime.ts | 13 ++++++---- src/signals/handlers.ts | 20 ++++++++------- src/signals/runtime.ts | 9 ++++--- src/signals/usage.ts | 14 +++++------ 12 files changed, 119 insertions(+), 83 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7b7ffa2..4b60539 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -34,10 +34,9 @@ order to create and drop others: no two test files share a database. See Three further commands are each their own CI step, kept out of the inner loop because each needs Docker or the network: -- **`npm run check:package`** packs the tarball, installs it into a throwaway project, and proves - every `exports` subpath resolves there for Node *and* the type checker. It is also the only thing - anywhere that holds the fourteen table names apart, and the only proof that the root still - resolves to nothing. +- **`npm run check:package`** builds the package tarball and installs it into a temporary project. + It checks that Node and the type checker can load every `exports` subpath. It is also the only + check that verifies all fourteen table names are unique and the package root cannot be imported. - **`npm run test:container`** is the single end-to-end test: a real container running real `pi` against the real Agent server, with a scripted model on localhost, so it needs no model credentials. About ten seconds. `CONCORDE_CONTAINER_TESTS` opts in. @@ -110,12 +109,12 @@ Four ways that fails quietly, all of them a deployment's to get right: types and yields no table and no schema object at all. The exception is eight names and no table among them — the four enum arrays and the unions derived from them, which a reader of a record has to be able to name. That overlap is permanent. -- **The schema object is `Schema` and the wrapper is `Tables`.** A name every - schema module declares carries a component prefix, because **`export *` drops a name that resolves - to more than one binding**: eight components exporting a bare `schema` yield a barrel exporting - none, `schemaFilter` derives to one schema, and `push` prints `Changes applied` having created one - component's tables. Two names are in that class and no more; the fourteen table names and the four - enum arrays are not prefixed. +- **Name the schema object `Schema` and the table wrapper `Tables`.** + The component prefix keeps these names unique when one module re-exports all schemas with + `export *`. If several components export the same name, such as `schema`, that name is left out + of the combined exports. Then `schemaFilter` selects only one schema, and `push` prints + `Changes applied` even though it created only that component's tables. Only the schema object + and table wrapper need these prefixes. The fourteen table names and four enum arrays do not. - **The table names are guarded by `scripts/check-package.ts`, which imports all eight `/schema` specifiers un-aliased into one module scope.** A duplicate is a `TS2300` and a `SyntaxError` naming both import lines. **An alias added to one of those lines removes the check for that diff --git a/CONTEXT.md b/CONTEXT.md index d7e3504..c863a30 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -129,7 +129,13 @@ Operator-defined JSON a Signal Handler writes on a Prompt beside the Session and _Avoid_: scope, claims, permissions, grant, ACL **Post phase**: -The Signal Handler's second entry point, used for cleanup. It runs once after all Runs from a Signal finish. It receives two flags: whether any Run failed and whether any Run exceeded its Usage limit. It cannot produce Prompts. +The Signal Handler's cleanup step. It runs once after all Runs for a Signal finish. It also runs if `handle` throws an error. This step cannot produce Prompts. + +It receives two values: + +- `failed`: true if `handle` or any Run failed. +- `limited`: true if any Run exceeded its Usage limit. + _Avoid_: post handler, teardown, finalizer, callback **Session**: @@ -141,16 +147,37 @@ One execution of the agent: a single Prompt, in one Session, producing whatever _Avoid_: turn, job, invocation, task **Usage**: -The model resources a Run uses, and the always-on infrastructure Component that reads the recorded usage. The Runtime reports usage for each model call, including retries and calls that compact the Session's history. The Signal Worker immediately saves each report as a **Usage record** linked to its Run. Saved records survive restarts. If a record cannot be saved, the Run stops to prevent further unrecorded usage. +The tokens a Run uses when it calls a model. Usage is also the name of the built-in Component that reads saved usage totals. This Component is always enabled. + +The Runtime reports usage after each model call. Retries count too, as do calls that shorten the Session's history by summarizing it. The Signal Worker saves each report as a **Usage record** for that Run as soon as it arrives. Saved records are kept after a restart. If saving fails, the Run stops so it does not keep using the model without recording usage. + +The Agent Implementation passes on the token counts reported by the model provider. Tokens do not have a single price: input, output, cache reads, and cache writes can each cost a different amount. + +The Usage Component can add up records for a chosen time period. It does not decide how much a User may use. The Operator writes that rule in a Signal Handler, including the budget, the time period it covers, and what to tell the User. -The Agent Implementation copies the provider's token counts. Token totals do not directly measure cost because providers price each token category differently. The framework can read totals for a time period, but does not enforce a budget from those totals. The Operator defines the budget, its time period, and messages to Users in a Signal Handler. The name "Usage" avoids confusion with "Token", which means a credential issued to a User. +We use the name "Usage" because "Token" already means the credential a User uses to authenticate. _Avoid_: Token Tracker, billing, metering, quota **Usage record**: -One saved row for one model call in a Run. It contains the provider, model, four token counts, their sum, and the cost calculated by the Agent Implementation. Records are written once and never updated. +A database row for one model call in a Run. It stores: + +- The provider and model used. +- The input, output, cache read, and cache write token counts. +- The total of those four counts. +- The cost calculated by the Agent Implementation. + +Each record is saved once and never changed. **Usage limit**: -The maximum model token count for one Run, set by a Signal Handler on the Prompt. The Runtime adds each call's usage when the call finishes. If the total exceeds the limit, it stops the agent and marks the Run as failed and limited. The Session keeps its history up to that point. A Run can exceed the limit by one call because usage is only known after a call finishes. This per-Run limit is the only limit the framework enforces. To enforce a budget, a Signal Handler passes the remaining token allowance. The limit counts every token category at full weight, including cache reads. Each model call sends the agent's whole context again and the provider reports most of it as cache reads, so cache reads usually make up most of a Run's count, although providers charge much less for them. +The token limit for one Run. A Signal Handler sets it on the Prompt as `usageLimit`. + +The Runtime adds up tokens as each model call finishes. If the total exceeds the limit, it stops the agent. The Run fails with `limited: true`. The Session keeps the history saved so far. + +Usage is only known after a model call finishes. This means the Run can go over the limit by one call. + +Every token counts equally toward this limit, including tokens read from the provider's cache. Each model call sends the agent's full context again. The provider reports most of that repeated context as cache reads. Cache reads therefore usually make up most of the token count, even though they cost much less. + +The framework only enforces a limit for each Run. To enforce a User's budget across Runs, the Signal Handler sets `usageLimit` to the number of tokens the User has left. _Avoid_: budget, quota, max tokens **Workspace**: diff --git a/README.md b/README.md index 709ff8c..b115def 100644 --- a/README.md +++ b/README.md @@ -71,7 +71,7 @@ from the registry rather than this checkout. | [Decisions](./examples/02_decisions/) | Two users and a shared record of signed decisions | | [Nostr](./examples/03_nostr/) | Encrypted direct messages through a local relay | | [Agent isolation](./examples/04_agent_isolation/) | Restricted agent networking and a model proxy that holds the credential | -| [Usage limits](./examples/05_usage_limits/) | Two users, each with a daily model usage limit | +| [Usage limits](./examples/05_usage_limits/) | Two users, each with a daily token limit | ## Deployment boundaries diff --git a/examples/05_usage_limits/AGENTS.md b/examples/05_usage_limits/AGENTS.md index 7aa72bc..4362a94 100644 --- a/examples/05_usage_limits/AGENTS.md +++ b/examples/05_usage_limits/AGENTS.md @@ -4,22 +4,23 @@ You help one user through a gateway. Keep answers brief. ## Gateway API -Use your shell and `curl` to reach `$AGENT_SERVER_URL`. No credential is required. -Read the live API description before calling it: +Use `curl` in the shell to call the API at `$AGENT_SERVER_URL`. You do not need a credential. +Read the current API description before making a request: ```sh curl -s "$AGENT_SERVER_URL/openapi.json" ``` -Use that document for request fields and error responses. If a call is refused, check the -response and correct the request before retrying. Do not repeatedly send the same invalid call. +The API description lists the request fields and error responses. If a request fails, read the +response and fix the request before trying again. Do not keep sending the same invalid request. ## Reply to the user Send replies with `POST /messages` and body `{"userId":"...","text":"..."}`. -Copy `userId` from the triggering signal; never invent one. Your final model response and +Copy `userId` from the Signal you are handling. Never invent one. Your final model response and workspace files are not delivered to the user. -Read prior messages with `GET /messages?user=`. The message log contains both directions; -use its cursors when paging. Write short, plain sentences without headings, tables, or code -blocks. The terminal client polls once a second, so delivery may follow your run. +Read earlier messages with `GET /messages?user=`. The log includes messages from both the +user and the agent. Use the returned cursors to request more pages. Write short, plain sentences +without headings, tables, or code blocks. The terminal client checks for messages once a second, +so the user may see your reply after your Run finishes. diff --git a/examples/05_usage_limits/README.md b/examples/05_usage_limits/README.md index de9e83b..b8b00e4 100644 --- a/examples/05_usage_limits/README.md +++ b/examples/05_usage_limits/README.md @@ -3,7 +3,7 @@ Alice and Bob talk to one agent over HTTP. Each user has a daily model token limit. When Alice reaches her limit, the gateway replies to her next message without calling the model. Bob can still get answers. This example defines the daily limit in `main.ts`. The framework records -Usage and stops individual Runs when they exceed their Usage limit. +Usage and stops each Run if it goes over its Usage limit. ## Requirements @@ -20,8 +20,9 @@ cp .env.example .env ``` Set `ANTHROPIC_API_KEY`, `ALICE_PASSWORD`, and `BOB_PASSWORD` in `.env`. `DAILY_TOKEN_LIMIT` is -the number of model tokens each user may use per UTC day. The value in `.env.example` is small -so you can reach the limit with a few messages. Check the model in `settings.json`. +the number of model tokens each user may use between midnight UTC and the next midnight UTC. +The value in `.env.example` is small so you can reach the limit with a few messages. Check the +model in `settings.json`. ```sh docker compose up -d --build @@ -50,8 +51,8 @@ before deciding whether to prompt the agent, so it does not use `templateHandler - `handle` calls `usage.total` to count the sender's tokens since 00:00 UTC today. If the total is at or above `DAILY_TOKEN_LIMIT`, it sends the limit message through the Messenger and returns no Prompt, so no Run starts. Otherwise, it returns one Prompt with the sender's - `userId` and sets `usageLimit` to their remaining daily token allowance. The Run's Usage is - added to that user's total. + `userId` and sets `usageLimit` to the number of tokens they have left for the day. The Run's + Usage is added to that user's total. - `post` receives the Run's outcome. If `limited` is true, the Run exceeded its Usage limit and the handler sends the limit message. For any other failure, it sends "I could not complete that request. Please try again in a minute." @@ -79,28 +80,31 @@ does not change the saved passwords, even though the example prints the `.env` v - **The counts are tokens, not money.** `DAILY_TOKEN_LIMIT` is compared with the total of input, output, cache read, and cache write tokens. Providers price those categories differently, so equal token totals can cost different amounts. -- **Most of the count is cache read tokens.** Every model call sends the agent's whole context - again: the system prompt, `AGENTS.md`, the tool definitions, and the conversation so far. The - provider serves most of that context from its cache and reports it as cache read tokens. A - reply is usually a few hundred output tokens, but one message can count 4,000 to 10,000 tokens - in total, and the count per message grows as the conversation gets longer. Providers usually - charge about a tenth of the normal input price for cache reads, but `DAILY_TOKEN_LIMIT` counts - them at full weight. Choose the limit with that in mind. To count the categories differently, - calculate your own number in the Signal Handler from the fields that `usage.total()` returns, - instead of using `totalTokens`. +- **Cached tokens count toward the limit.** Every model call sends the agent's full context + again: the system prompt, `AGENTS.md`, the tool definitions, and the conversation so far. + The provider reuses most of this context from its cache and reports those tokens as cache reads. + + A reply is usually a few hundred output tokens. But including the context, one message can use + 4,000 to 10,000 tokens in total. This total grows as the conversation gets longer. + + Cache reads usually cost about a tenth of the normal input price. Even so, each cached token + counts as one token toward `DAILY_TOKEN_LIMIT`. Account for these tokens when choosing a limit. + To give each token category a different weight, calculate the total in your Signal Handler + using the separate counts from `usage.total()`, rather than its `totalTokens` field. - **Each user has a separate limit.** Runs without a `userId` do not count toward any user's total. To set a hard spending cap for the whole deployment, enforce it at the model provider - or a model proxy that holds the credential, as in + or a model proxy that keeps the credential, as in [the isolation example](../04_agent_isolation/). ## Files and storage - `main.ts`: component setup, the daily limit handler, user creation, and shutdown. -- `AGENTS.md`: mounted agent instructions; `settings.json`: model configuration. +- `AGENTS.md`: instructions made available inside the agent container. +- `settings.json`: model configuration. - `compose.yml` and the Dockerfiles: gateway, pi RPC listener, database, migration, and two clients. -- `schema.ts`: tables for the four components. Usage records are part of the Signals schema, - so they need no separate schema setup. +- `schema.ts`: tables for the four components. The Signals schema already includes the Usage + table, so there is no extra schema to add. - `state/workspace/` and `state/sessions/`: persistent agent files and session records. Keep the agent server and RPC listener private. The [guide](../../site/guide.md) explains the diff --git a/scripts/check-package.ts b/scripts/check-package.ts index e666f6d..cbc855a 100644 --- a/scripts/check-package.ts +++ b/scripts/check-package.ts @@ -479,16 +479,15 @@ try { 'const own = pgSchema("consumer");', 'const notes = own.table("notes", { body: text("body").notNull() });', "", - "// And the framework's own, off the eight `/schema` subpaths, which sit one level below", - "// the component so that a config can name one as a", - "// path. Each table is imported by name at the top of this file rather than as a namespace,", - "// which is what states the shape: `drizzle-kit` reads `Object.values` of a module and keeps", - "// what passes `is(x, PgTable)`, so a table that had retreated into a `Tables`", - "// wrapper would be dropped in silence and generate an empty migration. Annotated as the two", - "// Drizzle types the exporter actually filters on, so a component that stopped exporting a", - "// table fails here rather than at an Operator's first `generate`. Nothing here is aliased:", - "// fourteen table names in one scope is a deployment's own barrel, and a fifteenth colliding", - "// with one of them is a compile error rather than a table missing from the DDL.", + "// Import tables by name from the eight component `/schema` subpaths. These paths let", + "// a deployment select schemas in its migration config. `drizzle-kit` reads the module's", + "// exports with `Object.values` and keeps values that pass `is(x, PgTable)`. It ignores", + "// tables inside a `Tables` object, which can leave a migration empty.", + "// The PgTable and PgSchema annotations below match the types Drizzle checks. If a", + "// component stops exporting a table, this check fails before a deployment runs `generate`.", + "// Import all fourteen table names without aliases, as a deployment's combined schema", + "// module would. A new table with a duplicate name then causes a compile error instead", + "// of being left out of the generated SQL.", "export const partTables: readonly PgTable[] = [", " signals, runs, usage, users, passwords, tokens,", " grants, admitted,", @@ -527,7 +526,7 @@ try { "// framework base type to import. It takes a `RunPrompt` and not a `Prompt`,", "// which is the seam's own type and the reason `prompt.session` is a string", "// here with nothing to check: the Signal Worker settled it.", - "// The worker always passes an observer, though runtimes may omit the parameter.", + "// The worker always passes an observer. Runtimes can choose whether to accept it.", "// Each report describes one model call and is saved by the worker as a Usage record.", "const runtime: Runtime = {", " async run(prompt: RunPrompt, observer?: RunObserver): Promise {", diff --git a/site/architecture.md b/site/architecture.md index 191a7b0..55e2216 100644 --- a/site/architecture.md +++ b/site/architecture.md @@ -28,14 +28,14 @@ The diagram shows messaging and authentication options. A messenger accepts one current implementation; the two channels shown are alternatives. The scheduler, signatures, and decisions can also be added, although they are not drawn. -`createGateway` constructs five infrastructure components: +`createGateway` creates five built-in components: | Component | Responsibility | | --- | --- | | Db | Shared PostgreSQL access so component state changes can commit together | | Agent server | HTTP capabilities the agent uses to read application data and act | | Public server | HTTP endpoints for user clients, including authentication and message exchange | -| Usage | Totals of the model usage the signal worker records for each run | +| Usage | Adds up the model usage saved by the signal worker for each run | | Signal worker | Durable queue that turns signals into agent runs, one signal at a time | Your `extend` callback constructs additional components using this infrastructure. Dependencies diff --git a/src/pi/output.ts b/src/pi/output.ts index e04d923..eb23595 100644 --- a/src/pi/output.ts +++ b/src/pi/output.ts @@ -10,9 +10,9 @@ type Answer = { export type Watch = { readonly observer?: RunObserver; - /** Ask the agent to stop, then keep reading until `agent_settled`. */ + /** Ask the agent to stop. The caller keeps reading until `agent_settled`. */ readonly abort: () => void; - /** Stop the Run and mark it as failed and limited if total tokens exceed this value. */ + /** Stop the agent and fail the Run with `limited: true` if token usage goes over this limit. */ readonly usageLimit?: number; }; @@ -82,7 +82,7 @@ export async function readOutcome( case "turn_end": answer = answerIn(record.message) ?? answer; break; - // Compaction uses a separate model call that is not included in message events. + // Summarizing the Session uses a separate model call, reported here instead of in a message. case "compaction_end": await report(usageIn(fieldsOf(record.result), named, true)); break; diff --git a/src/pi/runtime.ts b/src/pi/runtime.ts index 0027c40..128c582 100644 --- a/src/pi/runtime.ts +++ b/src/pi/runtime.ts @@ -23,11 +23,14 @@ const sessionNames = /^[A-Za-z0-9](?:[A-Za-z0-9._-]*[A-Za-z0-9])?$/; * include `.`, `_`, and `-`. Blank prompt text fails the run. Completion requires `agent_settled` * and a final assistant stop reason of `stop` or `length`. There is no execution timeout. * - * Reports usage for each assistant `message_end` and each `compaction_end`, including retries. - * If the observer throws or usage exceeds the Prompt's `usageLimit`, the runtime sends `abort` - * to pi. Once pi settles, the Run fails and the Session file is kept. Token counts come from - * the provider. A response interrupted mid-stream reports less output than was generated. - * If the server reports no usage, the recorded count is zero. + * Reports usage at each assistant `message_end`, including retries. Calls that summarize the + * Session's history report usage at `compaction_end`. If the observer throws an error or usage + * goes over the Prompt's `usageLimit`, the runtime sends `abort` to pi. It waits for + * `agent_settled`, then fails the Run and keeps the Session file. + * + * Token counts come from the provider. If a response is interrupted before it finishes, its + * reported output count is lower than the actual count. If the server reports no usage, the + * recorded count is zero. */ export function createPiRuntime(instance: PiInstance): Runtime { const { host, port, sessionsDir } = instance; diff --git a/src/signals/handlers.ts b/src/signals/handlers.ts index 93f7209..c3c65bc 100644 --- a/src/signals/handlers.ts +++ b/src/signals/handlers.ts @@ -18,18 +18,20 @@ export type Prompt = { */ readonly context?: unknown; /** - * The User whose usage totals include this Run. Use data authenticated by the Gateway, such - * as a Message's `userId`. Never use text written by a model or sender. The agent does not - * receive this field. It is separate from the Session name and prompt text. If omitted, - * the Run's usage is not assigned to any User. + * Add this Run's usage to this User's total. Take the id from data authenticated by the + * Gateway, such as a Message's `userId`. Never take it from text written by a model or sender. + * The agent does not receive this field. It is separate from the Session name and prompt text. + * Omit this field to leave the Run's usage unassigned to any User. */ readonly userId?: string; /** - * The maximum `totalTokens` for this Run, including compaction. If the total exceeds this - * limit, the runtime stops the agent and marks the Run as failed and limited. The Session - * keeps its history up to that point. Usage is reported after each model call finishes, so - * the Run can exceed the limit by at most one call. Must be a positive whole number. Omit - * this field for no limit. To enforce a budget, pass the remaining token allowance. + * The token limit for this Run, measured by `totalTokens`. This includes model calls that + * summarize the Session's history. If usage goes over the limit, the runtime stops the agent. + * The Run fails with `limited: true`, and the Session keeps its history so far. + * + * Usage is reported after each model call finishes, so the Run can go over the limit by at + * most one call. Use a positive whole number, or omit this field for no limit. To enforce a + * budget, pass the number of tokens left in that budget. */ readonly usageLimit?: number; }; diff --git a/src/signals/runtime.ts b/src/signals/runtime.ts index 8f2e2a0..d728f6f 100644 --- a/src/signals/runtime.ts +++ b/src/signals/runtime.ts @@ -18,7 +18,7 @@ export type UsageReport = { /** `null` when the provider is unknown, such as for some summarization calls. */ readonly provider: string | null; readonly model: string | null; - /** Prompt tokens not served from the provider's cache. */ + /** Prompt tokens that were not read from the provider's cache. */ readonly input: number; readonly output: number; readonly cacheRead: number; @@ -29,11 +29,12 @@ export type UsageReport = { readonly cost: number; }; -/** Supplied by the worker for one Run. */ +/** The worker provides this observer to receive usage reports for one Run. */ export type RunObserver = { /** - * The runtime calls and awaits this method once per model call, in order. If it throws, - * the runtime must stop and fail the Run to prevent further unrecorded usage. + * The runtime calls this method once for each model call, in order, and waits for it to finish. + * If it throws an error, the runtime must stop the Run and mark it as failed. The Run must + * not keep using the model if its usage cannot be saved. */ usage(report: UsageReport): void | Promise; }; diff --git a/src/signals/usage.ts b/src/signals/usage.ts index 231f0be..57cedcf 100644 --- a/src/signals/usage.ts +++ b/src/signals/usage.ts @@ -10,11 +10,11 @@ export type UsageOptions = { }; export type UsageQuery = { - /** Filter by User. Use `null` for Runs with no User, or omit to include all Users and Runs. */ + /** Select one User's records. Use `null` for Runs with no User. Omit to include all records. */ readonly userId?: string | null; /** Include records at or after this time. Omit to start from the first record. */ readonly since?: Date; - /** Exclusive. Omitted means up to now. */ + /** Include records before this time. Omit to leave the end time unrestricted. */ readonly until?: Date; }; @@ -30,13 +30,13 @@ export type UsageTotals = { }; /** - * Reads the Usage records saved by the Signal Worker, one per model call. Saved records - * survive restarts. Token counts come from the provider and do not directly measure cost. - * The accuracy of `cost` depends on the model price used by the agent implementation. - * `createGateway` constructs this component for you. + * Reads the Usage records saved by the Signal Worker, one per model call. Records are kept + * after a restart. Token counts come from the provider. Token counts alone do not tell you + * the cost: `cost` depends on the model price used by the agent implementation. + * `createGateway` creates this component for you. */ export type Usage = Component & { - /** Calculate totals in the database for records matching the User and time filters. */ + /** Add up matching records in the database, using the User and time filters. */ total(query?: UsageQuery): Promise; };