From 2946362cacf7dbc6a3e9c6752473400e39d5d0e7 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 19:58:28 +0000 Subject: [PATCH] feat(runtime): add durable next-event notice ledger (#99) Build the narrow notice core on the existing state kernel so recipient-scoped snapshots survive restarts and only evidence attempted next-event delivery. --- .changeset/durable-next-event-notices.md | 10 + docs/effect-conventions.md | 13 +- packages/rsc-runtime/README.md | 22 ++ packages/rsc-runtime/package.json | 4 + packages/rsc-runtime/rslib.config.ts | 13 + packages/rsc-runtime/src/agent-request.ts | 47 ++- packages/rsc-runtime/src/effect/boundary.ts | 1 + packages/rsc-runtime/src/index.ts | 9 + packages/rsc-runtime/src/notices/contract.ts | 154 ++++++++ packages/rsc-runtime/src/notices/index.ts | 50 +++ packages/rsc-runtime/src/notices/ledger.ts | 359 ++++++++++++++++++ packages/rsc-runtime/src/notices/state.ts | 264 +++++++++++++ packages/rsc-runtime/src/plugin.ts | 5 + .../tests/fixtures/notices-sqlite-process.mjs | 70 ++++ .../rsc-runtime/tests/notices-ledger.test.ts | 313 +++++++++++++++ .../notices-sqlite-cross-process.test.ts | 76 ++++ .../rsc-runtime/tests/state-packaging.test.ts | 27 +- rstest.integration-tests.ts | 1 + 18 files changed, 1424 insertions(+), 14 deletions(-) create mode 100644 .changeset/durable-next-event-notices.md create mode 100644 packages/rsc-runtime/src/notices/contract.ts create mode 100644 packages/rsc-runtime/src/notices/index.ts create mode 100644 packages/rsc-runtime/src/notices/ledger.ts create mode 100644 packages/rsc-runtime/src/notices/state.ts create mode 100644 packages/rsc-runtime/tests/fixtures/notices-sqlite-process.mjs create mode 100644 packages/rsc-runtime/tests/notices-ledger.test.ts create mode 100644 packages/rsc-runtime/tests/notices-sqlite-cross-process.test.ts diff --git a/.changeset/durable-next-event-notices.md b/.changeset/durable-next-event-notices.md new file mode 100644 index 000000000..213b01309 --- /dev/null +++ b/.changeset/durable-next-event-notices.md @@ -0,0 +1,10 @@ +--- +"@agent-bundle/runtime": minor +--- + +Add the optional recipient-scoped notice ledger behind the new `./notices` +subpath. It persists detached Agent Document snapshots through the existing +state kernel, exposes only the evidenced v1 states (`pending`, `attempted`, +`expired`, `unavailable`, `withdrawn`), performs publish- and delivery-time +authorization, and records next-event attempts with invocation receipts. +Stateless package-root and plugin consumers ship none of the ledger. diff --git a/docs/effect-conventions.md b/docs/effect-conventions.md index 4c2313652..79073c890 100644 --- a/docs/effect-conventions.md +++ b/docs/effect-conventions.md @@ -63,6 +63,7 @@ public or MCP-facing contracts. Internals keep the existing classes. | Success `A` | Promise resolves `A` | | Fail `AgentRequestError` | rethrow (`invalid-invocation`, `outside-invocation`, `request-closed`, `store-version-conflict`) | | Fail `AgentContractError` | rethrow (document / event / elapsed bounds) | +| Fail `AgentNoticeError` | rethrow (`aborted`, `invalid-input`, `request-closed`, `unauthorized`) | | Fail `AgentStateError` | rethrow. Matched by `error.name` in the root boundary so `./state/contract` never enters the package-root graph. | | Fail other `Error` | rethrow | | Fail non-Error | `new Error(String(value))` | @@ -157,6 +158,14 @@ Hurt / gotchas: Construct `Semaphore`/`Deferred` with the boundary's `runSync` and keep the admission bookkeeping synchronous. +## Stage 4 (#99 notice ledger) outcome + +The optional notice ledger composes state-kernel reads and dispatches inside +Effect programs, runs publish- and delivery-time authorization on the typed +error channel, and crosses back to the public Promise API only through the +runtime boundary. It adopts no unstable Effect modules and starts no fibers, +timers, or workers between invocations. + ## Banned modules and APIs - `Effect.runPromise` / `runSync` / `runFork` / `runCallback` (and `*With` / @@ -166,7 +175,7 @@ Hurt / gotchas: - `@effect/vitest` — this repo uses rstest. - `NodeRuntime.runMain` / `BunRuntime` as a substitute for the boundary. - Ad-hoc `ManagedRuntime` outside a boundary module. -- `effect/unstable/*` until listed below (Stages 2 and 3 listed none). +- `effect/unstable/*` until listed below (Stages 2, 3, and the #99 notice ledger listed none). ## Unstable-module adoptions @@ -174,6 +183,8 @@ Re-pin chores re-verify every row. Stage 2 adopts none: Flight is a React binary stream, not Ndjson/SchemaBinary, and no other `effect/unstable/*` module fits the dispatcher rewrite. Stage 3 also adopts none: the dev seam needed only stable `Semaphore`, `Deferred`, `Scope`, and `Exit`. +The #99 notice ledger also adopts none: it needs only stable `Effect` and +`forEach` over the existing Promise-returning state authority. | Module | Adopted in | Re-verify | | --- | --- | --- | diff --git a/packages/rsc-runtime/README.md b/packages/rsc-runtime/README.md index 3bec5b76a..8a1b24ac1 100644 --- a/packages/rsc-runtime/README.md +++ b/packages/rsc-runtime/README.md @@ -206,3 +206,25 @@ the test stand-in; it is never durable. The workspace-durable driver ships on including external ones — must pass the exported conformance suite (`stateDriverConformanceCases`); a disconnected adapter is not a completed integration. + +## Notices (optional) + +`@agent-bundle/runtime/notices` is the narrow recipient-scoped notice core. +It stores detached, finite `AgentDocumentSnapshot` content in one ordinary +state-kernel definition; host wiring opens that definition with the +workspace-durable SQLite driver and passes the resulting ledger as +`runAgentRequest({ noticeLedger })`. Stateless projects import neither +subpath and ship no state or notice implementation. + +Inside an authorized request, `(await agent()).notices` is a request-bound +handle with `publish()` and `read()`. Recipients use only observed +host/session/actor/workspace axes. Publish authorization runs before +persistence, and delivery authorization runs again when a matching event is +admitted. `read()` exposes notices selected for that event while the ledger +records a receipt containing the invocation id and state `attempted`. + +V1 deliberately exposes only `pending | attempted | expired | unavailable | +withdrawn`. It does not claim `delivered`, `read`, or `acknowledged`: observing +the recipient process is not evidence that the agent saw the content. There +is no router, MCP inbox, timer, retry worker, or autonomous work between +invocations in this subpath. diff --git a/packages/rsc-runtime/package.json b/packages/rsc-runtime/package.json index d6b7422c6..11cd97150 100644 --- a/packages/rsc-runtime/package.json +++ b/packages/rsc-runtime/package.json @@ -49,6 +49,10 @@ "./state/sqlite": { "types": "./dist/state/sqlite.d.ts", "import": "./dist/state/sqlite.js" + }, + "./notices": { + "types": "./dist/notices/index.d.ts", + "import": "./dist/notices.js" } }, "scripts": { diff --git a/packages/rsc-runtime/rslib.config.ts b/packages/rsc-runtime/rslib.config.ts index a98707593..780ad121c 100644 --- a/packages/rsc-runtime/rslib.config.ts +++ b/packages/rsc-runtime/rslib.config.ts @@ -20,6 +20,19 @@ export default defineConfig({ }, }, }, + { + ...sharedLib, + // Notices are optional and reuse the state entry's kernel runtime. + // Keeping this entry separate means stateless package-root consumers + // receive no ledger code and the notice entry never loads node:sqlite. + output: { + cleanDistPath: false, + externals: { '../state/index.js': './state.js' }, + }, + source: { + entry: { notices: './src/notices/index.ts' }, + }, + }, { ...sharedLib, // The sqlite driver is its own entry so `node:sqlite` (and its diff --git a/packages/rsc-runtime/src/agent-request.ts b/packages/rsc-runtime/src/agent-request.ts index 03ac0c132..6e310428a 100644 --- a/packages/rsc-runtime/src/agent-request.ts +++ b/packages/rsc-runtime/src/agent-request.ts @@ -1,9 +1,14 @@ import { AsyncLocalStorage } from 'node:async_hooks'; import type { JsonValue } from './lower-mcp.js'; +import type { + AgentNoticeLedger, + AgentNoticeRequestLease, + AgentNoticesHandle, +} from './notices/contract.js'; import type { AgentStateHandle } from './state/contract.js'; -export const AGENT_REQUEST_STORE_VERSION = 1; +export const AGENT_REQUEST_STORE_VERSION = 2; const STORE_SYMBOL = Symbol.for('@agent-bundle/runtime/request-store'); @@ -136,8 +141,12 @@ export interface AgentRequestContext { * `runAgentRequest({ state })`; undefined for stateless projects. */ readonly state: AgentStateHandle | undefined; - /** Reserved for recipient-aware notices (#99). Wave 1 leaves this undefined. */ - readonly notices: undefined; + /** + * Request-bound recipient notice handle (#99 narrow core). `read()` exposes + * notices attempted on this admitted event; `publish()` persists a detached + * Agent Document snapshot after publish-time authorization. + */ + readonly notices: AgentNoticesHandle | undefined; } export interface AgentRequestInit { @@ -145,6 +154,8 @@ export interface AgentRequestInit { readonly capabilities?: AgentRequestCapabilities; readonly host?: Observed; readonly invocation: AgentInvocationInput; + /** Optional durable notice authority; omitted projects load no notice code. */ + readonly noticeLedger?: AgentNoticeLedger; readonly progress?: AgentProgressReporter; readonly providers?: AgentProviderValues; readonly services?: AgentServiceRegistry; @@ -240,7 +251,7 @@ interface FrozenValues { readonly capabilities: AgentRequestCapabilities; readonly host: Observed; readonly invocation: AgentInvocation; - readonly notices: undefined; + readonly notices: AgentNoticesHandle | undefined; readonly progress: AgentProgressReporter; readonly providers: AgentProviderValues; readonly services: AgentServiceRegistry; @@ -348,19 +359,32 @@ export const runAgentRequest = async ( init: AgentRequestInit, operation: () => T | Promise, ): Promise => { + const actor = snapshotObserved(init.actor ?? unavailable()); + const host = snapshotObserved(init.host ?? unavailable()); + const invocation = invocationFrom(init.invocation); + const session = snapshotObserved(init.session ?? unavailable()); + const signal = init.signal ?? new AbortController().signal; + const workspace = snapshotObserved(init.workspace ?? unavailable()); + const noticeLease: AgentNoticeRequestLease | undefined = init.noticeLedger === undefined + ? undefined + : await init.noticeLedger.openRequest({ + invocation, + principal: Object.freeze({ actor, host, session, workspace }), + signal, + }); const values: FrozenValues = Object.freeze({ - actor: snapshotObserved(init.actor ?? unavailable()), + actor, capabilities: snapshotCapabilities(init.capabilities ?? emptyCapabilities()), - host: snapshotObserved(init.host ?? unavailable()), - invocation: invocationFrom(init.invocation), - notices: undefined, + host, + invocation, + notices: noticeLease?.handle, progress: init.progress ?? silentProgress, providers: Object.freeze({ ...(init.providers ?? {}) }), services: Object.freeze({ ...(init.services ?? {}) }), - session: snapshotObserved(init.session ?? unavailable()), - signal: init.signal ?? new AbortController().signal, + session, + signal, state: init.state, - workspace: snapshotObserved(init.workspace ?? unavailable()), + workspace, }); const lease: Lease = { closed: false, @@ -373,5 +397,6 @@ export const runAgentRequest = async ( return await getStore().storage.run(lease, operation); } finally { lease.closed = true; + noticeLease?.close(); } }; diff --git a/packages/rsc-runtime/src/effect/boundary.ts b/packages/rsc-runtime/src/effect/boundary.ts index f3c0ddb48..ccce1ae96 100644 --- a/packages/rsc-runtime/src/effect/boundary.ts +++ b/packages/rsc-runtime/src/effect/boundary.ts @@ -28,6 +28,7 @@ export interface RunPromiseOptions { const TYPED_ERROR_NAMES = new Set([ 'AgentContractError', + 'AgentNoticeError', 'AgentRequestError', 'AgentRuntimeError', 'AgentStateError', diff --git a/packages/rsc-runtime/src/index.ts b/packages/rsc-runtime/src/index.ts index 9764e6b02..0e645a035 100644 --- a/packages/rsc-runtime/src/index.ts +++ b/packages/rsc-runtime/src/index.ts @@ -88,5 +88,14 @@ export { createRscRequestContext } from './request-context.js'; // Type-only: the state kernel itself ships behind the './state' subpath so // stateless artifacts include none of it (#98). export type { AgentStateHandle, AgentStateLifetime } from './state/contract.js'; +// Type-only: the notice ledger itself ships behind the './notices' subpath. +export type { + AgentNotice, + AgentNoticeDelivery, + AgentNoticeLedger, + AgentNoticeState, + AgentNoticesHandle, + AgentRecipient, +} from './notices/contract.js'; export type { RscRequestContext } from './request-context.js'; export * from './plugin.js'; diff --git a/packages/rsc-runtime/src/notices/contract.ts b/packages/rsc-runtime/src/notices/contract.ts new file mode 100644 index 000000000..6ce771088 --- /dev/null +++ b/packages/rsc-runtime/src/notices/contract.ts @@ -0,0 +1,154 @@ +import type { AgentDocumentSnapshot } from '../agent-document.js'; +import type { + AgentActorIdentity, + AgentHostIdentity, + AgentInvocation, + AgentSessionIdentity, + AgentWorkspaceIdentity, + Observed, +} from '../agent-request.js'; + +export const AGENT_NOTICE_STATES = Object.freeze([ + 'pending', + 'attempted', + 'expired', + 'unavailable', + 'withdrawn', +] as const); + +/** V1 contains only states the framework can evidence without host claims. */ +export type AgentNoticeState = (typeof AGENT_NOTICE_STATES)[number]; + +export type AgentNoticePriority = 'low' | 'normal' | 'high'; + +/** A recipient is the conjunction of the observed identity axes it specifies. */ +export interface AgentRecipient { + readonly actor?: AgentActorIdentity; + readonly host?: AgentHostIdentity; + readonly session?: AgentSessionIdentity; + readonly workspace?: AgentWorkspaceIdentity; +} + +export interface AgentNoticePrincipal { + readonly actor: Observed; + readonly host: Observed; + readonly session: Observed; + readonly workspace: Observed; +} + +export interface AgentNoticeAttemptReceipt { + readonly attemptedAt: string; + readonly channel: 'next-event'; + readonly invocationId: string; +} + +export type AgentNoticeUnavailableReason = 'delivery-authorization-unavailable'; + +export interface AgentNotice { + readonly attempts: readonly AgentNoticeAttemptReceipt[]; + readonly content: AgentDocumentSnapshot; + readonly createdAt: string; + readonly dedupeKey?: string; + readonly expiredAt?: string; + readonly expiresAt?: string; + readonly id: string; + readonly priority: AgentNoticePriority; + readonly recipient: AgentRecipient; + readonly state: AgentNoticeState; + readonly unavailableAt?: string; + readonly unavailableReason?: AgentNoticeUnavailableReason; + readonly withdrawnAt?: string; +} + +export interface AgentNoticeLedgerSnapshot { + readonly notices: readonly AgentNotice[]; + readonly revision: number; +} + +export interface AgentNoticePublishInput { + readonly content: AgentDocumentSnapshot; + readonly dedupeKey?: string; + readonly expiresAt?: string; + readonly priority: AgentNoticePriority; + readonly recipient: AgentRecipient; +} + +export interface AgentNoticePublishOptions { + readonly idempotencyKey: string; +} + +export interface AgentNoticePublishResult { + readonly deduped: boolean; + readonly notice: AgentNotice; + readonly replayed: boolean; + readonly revision: number; +} + +export interface AgentNoticeExpiryOptions { + readonly at: string; + readonly idempotencyKey: string; +} + +export interface AgentNoticeWithdrawOptions { + readonly at: string; + readonly idempotencyKey: string; +} + +export type AgentNoticeAuthorizationDecision = + | { readonly state: 'authorized' } + | { readonly state: 'unavailable' }; + +export interface AgentNoticeAuthorizationRequest { + readonly noticeId?: string; + readonly phase: 'deliver' | 'publish'; + readonly principal: AgentNoticePrincipal; + readonly recipient: AgentRecipient; +} + +export type AgentNoticeAuthorizer = ( + request: AgentNoticeAuthorizationRequest, +) => AgentNoticeAuthorizationDecision | Promise; + +export interface AgentNoticeDelivery { + readonly notice: AgentNotice; + readonly receipt: AgentNoticeAttemptReceipt; +} + +export interface AgentNoticesHandle { + publish(input: AgentNoticePublishInput, options: AgentNoticePublishOptions): Promise; + read(): Promise; +} + +export interface AgentNoticeRequest { + readonly invocation: AgentInvocation; + readonly principal: AgentNoticePrincipal; + readonly signal: AbortSignal; +} + +export interface AgentNoticeRequestLease { + readonly handle: AgentNoticesHandle; + close(): void; +} + +export interface AgentNoticeLedger { + expire(options: AgentNoticeExpiryOptions): Promise; + openRequest(request: AgentNoticeRequest): Promise; + read(): Promise; + withdraw(id: string, options: AgentNoticeWithdrawOptions): Promise; +} + +export type AgentNoticeErrorCode = + | 'aborted' + | 'invalid-input' + | 'request-closed' + | 'unauthorized'; + +export class AgentNoticeError extends Error { + readonly code: AgentNoticeErrorCode; + + constructor(code: AgentNoticeErrorCode, message: string, options?: ErrorOptions) { + super(message, options); + this.code = code; + this.name = 'AgentNoticeError'; + } +} diff --git a/packages/rsc-runtime/src/notices/index.ts b/packages/rsc-runtime/src/notices/index.ts new file mode 100644 index 000000000..43ea1bea7 --- /dev/null +++ b/packages/rsc-runtime/src/notices/index.ts @@ -0,0 +1,50 @@ +/** + * Optional recipient-scoped notice ledger (#99 narrow core): + * `@agent-bundle/runtime/notices`. + * + * The ledger is implemented as one ordinary state-kernel definition. Host + * wiring chooses the matching driver; workspace durability therefore comes + * from `@agent-bundle/runtime/state/sqlite`, with the kernel owning + * transactions, revisions, and idempotency. + */ +export { + AGENT_NOTICE_STATES, + AgentNoticeError, +} from './contract.js'; +export type { + AgentNotice, + AgentNoticeAttemptReceipt, + AgentNoticeAuthorizationDecision, + AgentNoticeAuthorizationRequest, + AgentNoticeAuthorizer, + AgentNoticeDelivery, + AgentNoticeErrorCode, + AgentNoticeExpiryOptions, + AgentNoticeLedger, + AgentNoticeLedgerSnapshot, + AgentNoticePrincipal, + AgentNoticePriority, + AgentNoticePublishInput, + AgentNoticePublishOptions, + AgentNoticePublishResult, + AgentNoticeRequest, + AgentNoticeState, + AgentNoticesHandle, + AgentNoticeUnavailableReason, + AgentNoticeWithdrawOptions, + AgentRecipient, +} from './contract.js'; +export { + createAgentNoticeLedger, +} from './ledger.js'; +export type { + CreateAgentNoticeLedgerOptions, +} from './ledger.js'; +export { + agentNoticeEventSchemas, + agentNoticeStateDefinition, + recipientMatchesPrincipal, +} from './state.js'; +export type { + AgentNoticeLedgerState, +} from './state.js'; diff --git a/packages/rsc-runtime/src/notices/ledger.ts b/packages/rsc-runtime/src/notices/ledger.ts new file mode 100644 index 000000000..6bb4d4aed --- /dev/null +++ b/packages/rsc-runtime/src/notices/ledger.ts @@ -0,0 +1,359 @@ +import { createHash } from 'node:crypto'; + +import { Effect } from 'effect'; + +import { + createAgentDocument, + type AgentDocument, +} from '../agent-document.js'; +import { + runPromise, + toRuntimeError, +} from '../effect/boundary.js'; +import type { AgentStateStore } from '../state/contract.js'; +import { canonicalJson } from '../state/index.js'; +import { + AgentNoticeError, + type AgentNotice, + type AgentNoticeAuthorizationDecision, + type AgentNoticeAuthorizationRequest, + type AgentNoticeAuthorizer, + type AgentNoticeDelivery, + type AgentNoticeExpiryOptions, + type AgentNoticeLedger, + type AgentNoticeLedgerSnapshot, + type AgentNoticePrincipal, + type AgentNoticePublishInput, + type AgentNoticePublishOptions, + type AgentNoticePublishResult, + type AgentNoticeRequest, + type AgentNoticeRequestLease, + type AgentNoticesHandle, + type AgentNoticeWithdrawOptions, + type AgentRecipient, +} from './contract.js'; +import { + agentNoticeEventSchemas, + type AgentNoticeLedgerState, + recipientMatchesPrincipal, +} from './state.js'; + +export interface CreateAgentNoticeLedgerOptions { + readonly authorize: AgentNoticeAuthorizer; +} + +type NoticeStore = AgentStateStore; + +const noticeEffect = (evaluate: () => A): Effect.Effect => + Effect.try({ + catch: (error) => error, + try: evaluate, + }).pipe( + Effect.catch((error) => + error instanceof AgentNoticeError + ? Effect.fail(error) + : Effect.die(error), + ), + ); + +const storeEffect = (evaluate: () => Promise): Effect.Effect => + Effect.tryPromise({ + catch: toRuntimeError, + try: evaluate, + }); + +const nonEmptyText = (value: string, label: string): string => { + if (typeof value !== 'string' || value.trim() === '') { + throw new AgentNoticeError('invalid-input', `${label} must be a non-empty string`); + } + return value; +}; + +const timestamp = (value: string, label: string): string => { + nonEmptyText(value, label); + if (!Number.isFinite(Date.parse(value))) { + throw new AgentNoticeError('invalid-input', `${label} must be an ISO-8601 timestamp`); + } + return value; +}; + +const priority = (value: AgentNoticePublishInput['priority']): AgentNoticePublishInput['priority'] => { + switch (value) { + case 'low': + case 'normal': + case 'high': + return value; + default: { + const exhaustive: never = value; + throw new AgentNoticeError('invalid-input', `Unknown notice priority ${String(exhaustive)}`); + } + } +}; + +const recipient = (input: AgentRecipient): AgentRecipient => { + const result: AgentRecipient = Object.freeze({ + ...(input.actor === undefined + ? {} + : { actor: Object.freeze({ id: nonEmptyText(input.actor.id, 'Notice recipient actor id') }) }), + ...(input.host === undefined + ? {} + : { host: Object.freeze({ name: nonEmptyText(input.host.name, 'Notice recipient host name') }) }), + ...(input.session === undefined + ? {} + : { session: Object.freeze({ sessionId: nonEmptyText(input.session.sessionId, 'Notice recipient session id') }) }), + ...(input.workspace === undefined + ? {} + : { workspace: Object.freeze({ root: nonEmptyText(input.workspace.root, 'Notice recipient workspace root') }) }), + }); + if (Object.keys(result).length === 0) { + throw new AgentNoticeError('invalid-input', 'A notice recipient requires at least one observed identity axis'); + } + return result; +}; + +const authorizeEffect = ( + authorize: AgentNoticeAuthorizer, + request: AgentNoticeAuthorizationRequest, +): Effect.Effect => + Effect.tryPromise({ + catch: (error) => new AgentNoticeError( + 'unauthorized', + `Notice ${request.phase} authorization failed closed`, + { cause: error }, + ), + try: async () => authorize(request), + }).pipe( + Effect.flatMap((decision) => noticeEffect(() => { + switch (decision.state) { + case 'authorized': + case 'unavailable': + return decision; + default: { + const exhaustive: never = decision; + throw new AgentNoticeError( + 'unauthorized', + `Notice ${request.phase} authorization returned an invalid decision ${String(exhaustive)}`, + ); + } + } + })), + ); + +const findPublishedNotice = ( + notices: readonly AgentNotice[], + candidateId: string, + candidateRecipient: AgentRecipient, + dedupeKey: string | undefined, +): AgentNotice | undefined => notices.find((notice) => + notice.id === candidateId + || dedupeKey !== undefined + && notice.dedupeKey === dedupeKey + && canonicalJson(notice.recipient) === canonicalJson(candidateRecipient)); + +const snapshotFrom = ( + revision: number, + state: AgentNoticeLedgerState, +): AgentNoticeLedgerSnapshot => Object.freeze({ + notices: state.notices, + revision, +}); + +const assertOpen = (closed: boolean, signal: AbortSignal): void => { + if (closed) { + throw new AgentNoticeError('request-closed', 'Notice request handle used after the invocation completed'); + } + if (signal.aborted) { + throw new AgentNoticeError('aborted', 'Notice request operation was aborted', { cause: signal.reason }); + } +}; + +const publishProgram = ( + store: NoticeStore, + authorize: AgentNoticeAuthorizer, + request: AgentNoticeRequest, + input: AgentNoticePublishInput, + options: AgentNoticePublishOptions, +): Effect.Effect => Effect.gen(function*() { + const prepared = yield* noticeEffect(() => { + const target = recipient(input.recipient); + const createdAt = timestamp(request.invocation.startedAt, 'Notice createdAt'); + const expiresAt = input.expiresAt === undefined + ? undefined + : timestamp(input.expiresAt, 'Notice expiresAt'); + if (expiresAt !== undefined && Date.parse(expiresAt) <= Date.parse(createdAt)) { + throw new AgentNoticeError('invalid-input', 'Notice expiresAt must be later than createdAt'); + } + const idempotencyKey = nonEmptyText(options.idempotencyKey, 'Notice publish idempotency key'); + const dedupeKey = input.dedupeKey === undefined + ? undefined + : nonEmptyText(input.dedupeKey, 'Notice dedupe key'); + const id = `notice_${createHash('sha256') + .update(canonicalJson({ idempotencyKey, recipient: target }), 'utf8') + .digest('hex')}`; + const notice: AgentNotice = Object.freeze({ + attempts: Object.freeze([]), + content: createAgentDocument(input.content as AgentDocument), + createdAt, + ...(dedupeKey === undefined ? {} : { dedupeKey }), + ...(expiresAt === undefined ? {} : { expiresAt }), + id, + priority: priority(input.priority), + recipient: target, + state: 'pending', + }); + return { dedupeKey, id, idempotencyKey, notice, target }; + }); + const authorization = yield* authorizeEffect(authorize, { + phase: 'publish', + principal: request.principal, + recipient: prepared.target, + }); + if (authorization.state === 'unavailable') { + return yield* Effect.fail(new AgentNoticeError( + 'unauthorized', + 'Notice publish authorization is unavailable', + )); + } + const committed = yield* storeEffect(() => store.dispatch( + 'published', + { notice: prepared.notice }, + { + idempotencyKey: prepared.idempotencyKey, + signal: request.signal, + }, + )); + const persisted = findPublishedNotice( + committed.state.notices, + prepared.id, + prepared.target, + prepared.dedupeKey, + ); + if (persisted === undefined) { + return yield* Effect.die(new Error('Notice publish committed without a persisted notice')); + } + return Object.freeze({ + deduped: committed.replayed || persisted.id !== prepared.id, + notice: persisted, + replayed: committed.replayed, + revision: committed.revision, + }); +}); + +const deliveryFor = ( + notice: AgentNotice, + invocationId: string, +): AgentNoticeDelivery | undefined => { + if (notice.state !== 'attempted') return undefined; + const receipt = notice.attempts.find((attempt) => attempt.invocationId === invocationId); + return receipt === undefined ? undefined : Object.freeze({ notice, receipt }); +}; + +export const createAgentNoticeLedger = ( + store: NoticeStore, + options: CreateAgentNoticeLedgerOptions, +): AgentNoticeLedger => Object.freeze({ + expire(expiry: AgentNoticeExpiryOptions): Promise { + return runPromise(Effect.gen(function*() { + const at = yield* noticeEffect(() => timestamp(expiry.at, 'Notice expiry time')); + const idempotencyKey = yield* noticeEffect(() => + nonEmptyText(expiry.idempotencyKey, 'Notice expiry idempotency key')); + const committed = yield* storeEffect(() => store.dispatch( + 'expired', + { at }, + { idempotencyKey }, + )); + return snapshotFrom(committed.revision, committed.state); + })); + }, + + openRequest(request: AgentNoticeRequest): Promise { + return runPromise(Effect.gen(function*() { + let deliveries: readonly AgentNoticeDelivery[] = Object.freeze([]); + if (request.invocation.kind === 'event') { + const before = yield* storeEffect(() => store.read({ signal: request.signal })); + const admissionTime = Date.parse(request.invocation.startedAt); + const expiring = before.state.notices.filter((notice) => + notice.state === 'pending' + && notice.expiresAt !== undefined + && Date.parse(notice.expiresAt) <= admissionTime); + const candidates = before.state.notices.filter((notice) => + notice.state === 'pending' + && (notice.expiresAt === undefined + || Date.parse(notice.expiresAt) > admissionTime) + && recipientMatchesPrincipal(notice.recipient, request.principal)); + const decisions = yield* Effect.forEach(candidates, (notice) => + authorizeEffect(options.authorize, { + noticeId: notice.id, + phase: 'deliver', + principal: request.principal, + recipient: notice.recipient, + }).pipe(Effect.map((decision) => ({ decision, id: notice.id })))); + if (expiring.length > 0 || decisions.length > 0) { + const committed = yield* storeEffect(() => store.dispatch( + 'admitted', + { + at: request.invocation.startedAt, + authorizedIds: decisions + .filter(({ decision }) => decision.state === 'authorized') + .map(({ id }) => id), + invocationId: request.invocation.id, + principal: request.principal, + unavailableIds: decisions + .filter(({ decision }) => decision.state === 'unavailable') + .map(({ id }) => id), + }, + { + idempotencyKey: `agent-notices:admit:${request.invocation.id}`, + signal: request.signal, + }, + )); + deliveries = Object.freeze(committed.state.notices + .map((notice) => deliveryFor(notice, request.invocation.id)) + .filter((delivery): delivery is AgentNoticeDelivery => delivery !== undefined)); + } + } + + let closed = false; + const handle: AgentNoticesHandle = Object.freeze({ + publish(input: AgentNoticePublishInput, publishOptions: AgentNoticePublishOptions) { + return runPromise(Effect.gen(function*() { + yield* noticeEffect(() => assertOpen(closed, request.signal)); + return yield* publishProgram(store, options.authorize, request, input, publishOptions); + })); + }, + read() { + return runPromise(noticeEffect(() => { + assertOpen(closed, request.signal); + return deliveries; + })); + }, + }); + return Object.freeze({ + close() { + closed = true; + }, + handle, + }); + })); + }, + + async read(): Promise { + const snapshot = await store.read(); + return snapshotFrom(snapshot.revision, snapshot.state); + }, + + withdraw(id: string, withdrawal: AgentNoticeWithdrawOptions): Promise { + return runPromise(Effect.gen(function*() { + const noticeId = yield* noticeEffect(() => nonEmptyText(id, 'Notice id')); + const at = yield* noticeEffect(() => timestamp(withdrawal.at, 'Notice withdrawal time')); + const idempotencyKey = yield* noticeEffect(() => + nonEmptyText(withdrawal.idempotencyKey, 'Notice withdrawal idempotency key')); + const committed = yield* storeEffect(() => store.dispatch( + 'withdrawn', + { at, id: noticeId }, + { idempotencyKey }, + )); + return snapshotFrom(committed.revision, committed.state); + })); + }, +}); diff --git a/packages/rsc-runtime/src/notices/state.ts b/packages/rsc-runtime/src/notices/state.ts new file mode 100644 index 000000000..6abc7cbb9 --- /dev/null +++ b/packages/rsc-runtime/src/notices/state.ts @@ -0,0 +1,264 @@ +import { z } from 'zod'; + +import { + createAgentDocument, + type AgentDocument, + type AgentDocumentSnapshot, +} from '../agent-document.js'; +import type { + AgentStateDefinition, + AgentStateEventSchemas, + AgentStateLifetime, +} from '../state/contract.js'; +import { canonicalJson, defineState } from '../state/index.js'; +import type { + AgentNotice, + AgentNoticePrincipal, + AgentRecipient, +} from './contract.js'; + +const observed = (value: T) => z.discriminatedUnion('state', [ + z.object({ + source: z.enum(['native', 'receipt', 'derived']), + state: z.literal('available'), + value, + }).strict(), + z.object({ + reason: z.enum(['not-provided', 'unsupported-surface', 'host-omitted', 'unauthenticated']), + state: z.literal('unavailable'), + }).strict(), +]); + +const recipientSchema = z.object({ + actor: z.object({ id: z.string().min(1) }).strict().optional(), + host: z.object({ name: z.string().min(1) }).strict().optional(), + session: z.object({ sessionId: z.string().min(1) }).strict().optional(), + workspace: z.object({ root: z.string().min(1) }).strict().optional(), +}).strict().refine( + (recipient) => Object.values(recipient).some((value) => value !== undefined), + 'A notice recipient requires at least one identity axis', +); + +const principalSchema = z.object({ + actor: observed(z.object({ id: z.string().min(1) }).strict()), + host: observed(z.object({ name: z.string().min(1) }).strict()), + session: observed(z.object({ sessionId: z.string().min(1) }).strict()), + workspace: observed(z.object({ root: z.string().min(1) }).strict()), +}).strict(); + +const documentSchema = z.custom((value) => { + try { + createAgentDocument(value as AgentDocument); + return true; + } catch { + return false; + } +}); + +const attemptSchema = z.object({ + attemptedAt: z.string().min(1), + channel: z.literal('next-event'), + invocationId: z.string().min(1), +}).strict().readonly(); + +const noticeSchema = z.object({ + attempts: z.array(attemptSchema).readonly(), + content: documentSchema, + createdAt: z.string().min(1), + dedupeKey: z.string().min(1).optional(), + expiredAt: z.string().min(1).optional(), + expiresAt: z.string().min(1).optional(), + id: z.string().min(1), + priority: z.enum(['low', 'normal', 'high']), + recipient: recipientSchema, + state: z.enum(['pending', 'attempted', 'expired', 'unavailable', 'withdrawn']), + unavailableAt: z.string().min(1).optional(), + unavailableReason: z.literal('delivery-authorization-unavailable').optional(), + withdrawnAt: z.string().min(1).optional(), +}).strict().readonly(); + +export interface AgentNoticeLedgerState { + readonly notices: readonly AgentNotice[]; +} + +export const agentNoticeEventSchemas = { + admitted: z.object({ + at: z.string().min(1), + authorizedIds: z.array(z.string().min(1)), + invocationId: z.string().min(1), + principal: principalSchema, + unavailableIds: z.array(z.string().min(1)), + }).strict(), + expired: z.object({ + at: z.string().min(1), + }).strict(), + published: z.object({ + notice: noticeSchema, + }).strict(), + withdrawn: z.object({ + at: z.string().min(1), + id: z.string().min(1), + }).strict(), +} as const satisfies AgentStateEventSchemas; + +const sameRecipient = (left: AgentRecipient, right: AgentRecipient): boolean => + canonicalJson(left) === canonicalJson(right); + +export const recipientMatchesPrincipal = ( + recipient: AgentRecipient, + principal: AgentNoticePrincipal, +): boolean => { + if (recipient.actor !== undefined) { + if (principal.actor.state !== 'available' || principal.actor.value.id !== recipient.actor.id) return false; + } + if (recipient.host !== undefined) { + if (principal.host.state !== 'available' || principal.host.value.name !== recipient.host.name) return false; + } + if (recipient.session !== undefined) { + if (principal.session.state !== 'available' || principal.session.value.sessionId !== recipient.session.sessionId) return false; + } + if (recipient.workspace !== undefined) { + if (principal.workspace.state !== 'available' || principal.workspace.value.root !== recipient.workspace.root) return false; + } + return true; +}; + +const expiredNotice = (notice: AgentNotice, at: string): AgentNotice => Object.freeze({ + ...notice, + expiredAt: at, + state: 'expired', +}); + +const transitionExpiry = (notice: AgentNotice, at: string): AgentNotice => { + switch (notice.state) { + case 'pending': + return notice.expiresAt !== undefined && Date.parse(notice.expiresAt) <= Date.parse(at) + ? expiredNotice(notice, at) + : notice; + case 'attempted': + case 'expired': + case 'unavailable': + case 'withdrawn': + return notice; + default: { + const exhaustive: never = notice.state; + return exhaustive; + } + } +}; + +const transitionWithdrawal = (notice: AgentNotice, id: string, at: string): AgentNotice => { + switch (notice.state) { + case 'pending': + return notice.id === id + ? Object.freeze({ ...notice, state: 'withdrawn', withdrawnAt: at }) + : notice; + case 'attempted': + case 'expired': + case 'unavailable': + case 'withdrawn': + return notice; + default: { + const exhaustive: never = notice.state; + return exhaustive; + } + } +}; + +const transitionAdmission = ( + notice: AgentNotice, + input: { + readonly at: string; + readonly authorizedIds: ReadonlySet; + readonly invocationId: string; + readonly principal: AgentNoticePrincipal; + readonly unavailableIds: ReadonlySet; + }, +): AgentNotice => { + const current = transitionExpiry(notice, input.at); + switch (current.state) { + case 'pending': { + if (!recipientMatchesPrincipal(current.recipient, input.principal)) return current; + if (input.unavailableIds.has(current.id)) { + return Object.freeze({ + ...current, + state: 'unavailable', + unavailableAt: input.at, + unavailableReason: 'delivery-authorization-unavailable', + }); + } + if (!input.authorizedIds.has(current.id)) return current; + const receipt = Object.freeze({ + attemptedAt: input.at, + channel: 'next-event' as const, + invocationId: input.invocationId, + }); + return Object.freeze({ + ...current, + attempts: Object.freeze([...current.attempts, receipt]), + state: 'attempted', + }); + } + case 'attempted': + case 'expired': + case 'unavailable': + case 'withdrawn': + return current; + default: { + const exhaustive: never = current.state; + return exhaustive; + } + } +}; + +export const agentNoticeStateDefinition = ( + lifetime: AgentStateLifetime = 'workspace-durable', +): AgentStateDefinition => + defineState({ + events: agentNoticeEventSchemas, + id: '@agent-bundle/runtime/agent-notice-ledger/v1', + initial: { notices: [] }, + lifetime, + reduce: (state, event) => { + switch (event.name) { + case 'published': { + const candidate = event.payload.notice; + const duplicate = candidate.dedupeKey === undefined + ? state.notices.some((notice) => notice.id === candidate.id) + : state.notices.some((notice) => + notice.dedupeKey === candidate.dedupeKey + && sameRecipient(notice.recipient, candidate.recipient)); + return duplicate ? state : { notices: [...state.notices, candidate] }; + } + case 'expired': + return { + notices: state.notices.map((notice) => transitionExpiry(notice, event.payload.at)), + }; + case 'withdrawn': + return { + notices: state.notices.map((notice) => + transitionWithdrawal(notice, event.payload.id, event.payload.at)), + }; + case 'admitted': { + const authorizedIds = new Set(event.payload.authorizedIds); + const unavailableIds = new Set(event.payload.unavailableIds); + return { + notices: state.notices.map((notice) => transitionAdmission(notice, { + at: event.payload.at, + authorizedIds, + invocationId: event.payload.invocationId, + principal: event.payload.principal, + unavailableIds, + })), + }; + } + default: { + const exhaustive: never = event; + return exhaustive; + } + } + }, + schema: z.object({ + notices: z.array(noticeSchema).readonly(), + }).strict().readonly(), + }); diff --git a/packages/rsc-runtime/src/plugin.ts b/packages/rsc-runtime/src/plugin.ts index e99f88442..0adee97db 100644 --- a/packages/rsc-runtime/src/plugin.ts +++ b/packages/rsc-runtime/src/plugin.ts @@ -36,6 +36,11 @@ export type { Observed, ObservedSource, } from './agent-request.js'; +// Type-only: the optional ledger implementation stays behind './notices'. +export type { + AgentNoticeLedger, + AgentNoticesHandle, +} from './notices/contract.js'; export { defineRscApplication } from './application.js'; export type { RscApplication, RscApplicationOptions } from './application.js'; export { runRscCli } from './cli.js'; diff --git a/packages/rsc-runtime/tests/fixtures/notices-sqlite-process.mjs b/packages/rsc-runtime/tests/fixtures/notices-sqlite-process.mjs new file mode 100644 index 000000000..6be09a180 --- /dev/null +++ b/packages/rsc-runtime/tests/fixtures/notices-sqlite-process.mjs @@ -0,0 +1,70 @@ +import { + agent, + available, + runAgentRequest, +} from '../../dist/index.js'; +import { + agentNoticeStateDefinition, + createAgentNoticeLedger, +} from '../../dist/notices.js'; +import { createSqliteStateDriver } from '../../dist/state/sqlite.js'; + +const [file, mode] = process.argv.slice(2); +if (typeof file !== 'string' || (mode !== 'publish' && mode !== 'deliver')) { + throw new Error('usage: notices-sqlite-process.mjs '); +} + +const driver = createSqliteStateDriver({ file }); +const store = await driver.open(agentNoticeStateDefinition()); +const ledger = createAgentNoticeLedger(store, { + authorize: () => ({ state: 'authorized' }), +}); + +try { + if (mode === 'publish') { + const result = await runAgentRequest({ + actor: available({ id: 'publisher' }, 'native'), + host: available({ name: 'claude' }, 'native'), + invocation: { + id: 'publish-process', + kind: 'tool', + startedAt: '2026-09-01T19:00:00.000Z', + }, + noticeLedger: ledger, + session: available({ sessionId: 'session-1' }, 'native'), + workspace: available({ root: '/workspace' }, 'native'), + }, async () => (await agent()).notices.publish({ + content: { + root: { kind: 'text', text: 'cross-process notice' }, + status: 'success', + version: 1, + }, + dedupeKey: 'cross-process', + priority: 'high', + recipient: { + actor: { id: 'recipient' }, + workspace: { root: '/workspace' }, + }, + }, { idempotencyKey: 'publish:cross-process' })); + process.stdout.write(JSON.stringify({ + id: result.notice.id, + state: result.notice.state, + })); + } else { + const result = await runAgentRequest({ + actor: available({ id: 'recipient' }, 'native'), + host: available({ name: 'claude' }, 'native'), + invocation: { + id: 'delivery-process', + kind: 'event', + startedAt: '2026-09-01T19:05:00.000Z', + }, + noticeLedger: ledger, + session: available({ sessionId: 'session-1' }, 'native'), + workspace: available({ root: '/workspace' }, 'native'), + }, async () => (await agent()).notices.read()); + process.stdout.write(JSON.stringify(result)); + } +} finally { + await driver.close(); +} diff --git a/packages/rsc-runtime/tests/notices-ledger.test.ts b/packages/rsc-runtime/tests/notices-ledger.test.ts new file mode 100644 index 000000000..6d4fd0ac3 --- /dev/null +++ b/packages/rsc-runtime/tests/notices-ledger.test.ts @@ -0,0 +1,313 @@ +import { describe, expect, it } from '@rstest/core'; + +import { + AGENT_NOTICE_STATES, + AgentNoticeError, + agentNoticeStateDefinition, + createAgentNoticeLedger, + type AgentNoticeAuthorizationRequest, + type AgentNoticeState, +} from '../src/notices/index.js'; +import { + agent, + available, + runAgentRequest, +} from '../src/index.js'; +import { createMemoryStateDriver } from '../src/state/index.js'; + +const document = (text: string) => ({ + root: { kind: 'text' as const, text }, + status: 'success' as const, + version: 1 as const, +}); + +const actor = (id: string) => available({ id }, 'native'); +const host = available({ name: 'claude' }, 'native'); +const session = available({ sessionId: 'session-1' }, 'native'); +const workspace = available({ root: '/workspace' }, 'native'); + +const openLedger = async ( + authorize: (request: AgentNoticeAuthorizationRequest) => + | { readonly state: 'authorized' | 'unavailable' } + | Promise<{ readonly state: 'authorized' | 'unavailable' }> = + () => ({ state: 'authorized' }), +) => { + const driver = createMemoryStateDriver({ lifetime: 'process' }); + const store = await driver.open(agentNoticeStateDefinition('process')); + const ledger = createAgentNoticeLedger(store, { authorize }); + return { driver, ledger }; +}; + +const run = async ( + ledger: Awaited>['ledger'], + input: { + readonly actorId: string; + readonly id: string; + readonly kind: 'event' | 'tool'; + readonly startedAt: string; + }, + operation: () => Promise, +): Promise => runAgentRequest({ + actor: actor(input.actorId), + host, + invocation: { + id: input.id, + kind: input.kind, + startedAt: input.startedAt, + }, + noticeLedger: ledger, + session, + workspace, +}, operation); + +describe('notice state taxonomy', () => { + it('declares only framework-evidenced v1 states', () => { + expect(AGENT_NOTICE_STATES).toEqual([ + 'pending', + 'attempted', + 'expired', + 'unavailable', + 'withdrawn', + ]); + expect(AGENT_NOTICE_STATES).not.toContain('delivered'); + expect(AGENT_NOTICE_STATES).not.toContain('read'); + expect(AGENT_NOTICE_STATES).not.toContain('acknowledged'); + + const label = (state: AgentNoticeState): string => { + switch (state) { + case 'pending': + case 'attempted': + case 'expired': + case 'unavailable': + case 'withdrawn': + return state; + default: { + const exhaustive: never = state; + return exhaustive; + } + } + }; + expect(AGENT_NOTICE_STATES.map(label)).toEqual(AGENT_NOTICE_STATES); + }); +}); + +describe('durable notice ledger', () => { + it('publishes a detached finite snapshot and dedupes atomically by recipient key', async () => { + const { driver, ledger } = await openLedger(); + const source = document('original'); + const first = await run(ledger, { + actorId: 'publisher', + id: 'publish-1', + kind: 'tool', + startedAt: '2026-09-01T19:00:00.000Z', + }, async () => (await agent()).notices!.publish({ + content: source, + dedupeKey: 'conflict:file-a', + priority: 'high', + recipient: { + actor: { id: 'recipient' }, + workspace: { root: '/workspace' }, + }, + }, { idempotencyKey: 'publish:1' })); + + (source.root as { text: string }).text = 'mutated'; + const second = await run(ledger, { + actorId: 'publisher', + id: 'publish-2', + kind: 'tool', + startedAt: '2026-09-01T19:01:00.000Z', + }, async () => (await agent()).notices!.publish({ + content: document('duplicate content is not persisted'), + dedupeKey: 'conflict:file-a', + priority: 'low', + recipient: { + actor: { id: 'recipient' }, + workspace: { root: '/workspace' }, + }, + }, { idempotencyKey: 'publish:2' })); + + expect(first).toMatchObject({ deduped: false, replayed: false }); + expect(second).toMatchObject({ deduped: true, notice: { id: first.notice.id } }); + expect((await ledger.read()).notices).toHaveLength(1); + expect(first.notice.content.root).toEqual({ kind: 'text', text: 'original' }); + expect(Object.isFrozen(first.notice)).toBe(true); + expect(Object.isFrozen(first.notice.content)).toBe(true); + expect(Object.isFrozen(first.notice.content.root)).toBe(true); + await driver.close(); + }); + + it('expires and withdraws pending notices without inventing stronger states', async () => { + const { driver, ledger } = await openLedger(); + const expiring = await run(ledger, { + actorId: 'publisher', + id: 'publish-expiring', + kind: 'tool', + startedAt: '2026-09-01T19:00:00.000Z', + }, async () => (await agent()).notices!.publish({ + content: document('expires'), + expiresAt: '2026-09-01T19:05:00.000Z', + priority: 'normal', + recipient: { actor: { id: 'recipient' } }, + }, { idempotencyKey: 'publish:expiring' })); + const withdrawn = await run(ledger, { + actorId: 'publisher', + id: 'publish-withdrawn', + kind: 'tool', + startedAt: '2026-09-01T19:00:01.000Z', + }, async () => (await agent()).notices!.publish({ + content: document('withdrawn'), + priority: 'normal', + recipient: { actor: { id: 'recipient' } }, + }, { idempotencyKey: 'publish:withdrawn' })); + + await ledger.withdraw(withdrawn.notice.id, { + at: '2026-09-01T19:03:00.000Z', + idempotencyKey: 'withdraw:1', + }); + await ledger.expire({ + at: '2026-09-01T19:06:00.000Z', + idempotencyKey: 'expire:1', + }); + + const snapshot = await ledger.read(); + expect(snapshot.notices.find((notice) => notice.id === expiring.notice.id)?.state).toBe('expired'); + expect(snapshot.notices.find((notice) => notice.id === withdrawn.notice.id)?.state).toBe('withdrawn'); + await driver.close(); + }); + + it('fails publish authorization before persistence', async () => { + const phases: string[] = []; + const { driver, ledger } = await openLedger((request) => { + phases.push(request.phase); + return { state: 'unavailable' }; + }); + + await expect(run(ledger, { + actorId: 'publisher', + id: 'publish-denied', + kind: 'tool', + startedAt: '2026-09-01T19:00:00.000Z', + }, async () => (await agent()).notices!.publish({ + content: document('denied'), + priority: 'normal', + recipient: { actor: { id: 'recipient' } }, + }, { idempotencyKey: 'publish:denied' }))).rejects.toBeInstanceOf(AgentNoticeError); + expect(phases).toEqual(['publish']); + expect((await ledger.read()).notices).toEqual([]); + await driver.close(); + }); +}); + +describe('next-event delivery', () => { + it('exposes only matching pending notices and records an attempted receipt', async () => { + const phases: string[] = []; + const { driver, ledger } = await openLedger((request) => { + phases.push(request.phase); + return { state: 'authorized' }; + }); + const published = await run(ledger, { + actorId: 'publisher', + id: 'publish', + kind: 'tool', + startedAt: '2026-09-01T19:00:00.000Z', + }, async () => (await agent()).notices!.publish({ + content: document('next event'), + priority: 'high', + recipient: { + actor: { id: 'recipient' }, + host: { name: 'claude' }, + session: { sessionId: 'session-1' }, + workspace: { root: '/workspace' }, + }, + }, { idempotencyKey: 'publish:next' })); + + const wrongActor = await run(ledger, { + actorId: 'other', + id: 'event-other', + kind: 'event', + startedAt: '2026-09-01T19:01:00.000Z', + }, async () => (await agent()).notices!.read()); + expect(wrongActor).toEqual([]); + expect(await ledger.read()).toMatchObject({ + notices: [expect.objectContaining({ state: 'pending' })], + revision: 1, + }); + + const observed = await run(ledger, { + actorId: 'recipient', + id: 'event-recipient', + kind: 'event', + startedAt: '2026-09-01T19:02:00.000Z', + }, async () => (await agent()).notices!.read()); + expect(observed).toEqual([expect.objectContaining({ + notice: expect.objectContaining({ + id: published.notice.id, + state: 'attempted', + }), + receipt: { + attemptedAt: '2026-09-01T19:02:00.000Z', + channel: 'next-event', + invocationId: 'event-recipient', + }, + })]); + expect(phases).toEqual(['publish', 'deliver']); + expect((await ledger.read()).notices[0]).toMatchObject({ + attempts: [{ invocationId: 'event-recipient' }], + state: 'attempted', + }); + await driver.close(); + }); + + it('marks a matched notice unavailable when delivery-time authorization is unavailable', async () => { + const { driver, ledger } = await openLedger((request) => ({ + state: request.phase === 'publish' ? 'authorized' : 'unavailable', + })); + await run(ledger, { + actorId: 'publisher', + id: 'publish', + kind: 'tool', + startedAt: '2026-09-01T19:00:00.000Z', + }, async () => (await agent()).notices!.publish({ + content: document('not authorized later'), + priority: 'normal', + recipient: { actor: { id: 'recipient' } }, + }, { idempotencyKey: 'publish:unavailable' })); + + const observed = await run(ledger, { + actorId: 'recipient', + id: 'event-recipient', + kind: 'event', + startedAt: '2026-09-01T19:02:00.000Z', + }, async () => (await agent()).notices!.read()); + expect(observed).toEqual([]); + expect((await ledger.read()).notices[0]).toMatchObject({ + state: 'unavailable', + unavailableAt: '2026-09-01T19:02:00.000Z', + unavailableReason: 'delivery-authorization-unavailable', + }); + await driver.close(); + }); + + it('does not attempt delivery during non-event invocations', async () => { + const { driver, ledger } = await openLedger(); + await run(ledger, { + actorId: 'publisher', + id: 'publish', + kind: 'tool', + startedAt: '2026-09-01T19:00:00.000Z', + }, async () => (await agent()).notices!.publish({ + content: document('wait'), + priority: 'normal', + recipient: { actor: { id: 'recipient' } }, + }, { idempotencyKey: 'publish:wait' })); + + expect(await run(ledger, { + actorId: 'recipient', + id: 'tool-recipient', + kind: 'tool', + startedAt: '2026-09-01T19:02:00.000Z', + }, async () => (await agent()).notices!.read())).toEqual([]); + expect((await ledger.read()).notices[0]?.state).toBe('pending'); + await driver.close(); + }); +}); diff --git a/packages/rsc-runtime/tests/notices-sqlite-cross-process.test.ts b/packages/rsc-runtime/tests/notices-sqlite-cross-process.test.ts new file mode 100644 index 000000000..0b60f8be9 --- /dev/null +++ b/packages/rsc-runtime/tests/notices-sqlite-cross-process.test.ts @@ -0,0 +1,76 @@ +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; + +import { describe, expect, it } from '@rstest/core'; + +import { + agentNoticeStateDefinition, + createAgentNoticeLedger, +} from '../src/notices/index.js'; +import { createSqliteStateDriver } from '../src/state/sqlite.js'; + +const packageRoot = fileURLToPath(new URL('..', import.meta.url)); +const fixture = join(packageRoot, 'tests', 'fixtures', 'notices-sqlite-process.mjs'); + +const runProcess = async ( + file: string, + mode: 'deliver' | 'publish', +): Promise => { + const child = spawn(process.execPath, [fixture, file, mode], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + const stdout: Buffer[] = []; + const stderr: Buffer[] = []; + child.stdout.on('data', (chunk: Buffer) => stdout.push(chunk)); + child.stderr.on('data', (chunk: Buffer) => stderr.push(chunk)); + const [code] = await once(child, 'close') as [number | null]; + expect(code, Buffer.concat(stderr).toString('utf8')).toBe(0); + return JSON.parse(Buffer.concat(stdout).toString('utf8')); +}; + +describe.sequential('notice ledger cross-process proof', () => { + it('publishes in one process and attempts delivery on the next admitted event in another', { timeout: 60_000 }, async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-notices-cross-process-')); + const file = join(root, 'notices.sqlite'); + try { + const published = await runProcess(file, 'publish') as { readonly id: string; readonly state: string }; + expect(published.state).toBe('pending'); + + const observed = await runProcess(file, 'deliver') as readonly [{ + readonly notice: { readonly id: string; readonly state: string }; + readonly receipt: { readonly channel: string; readonly invocationId: string }; + }]; + expect(observed).toEqual([expect.objectContaining({ + notice: expect.objectContaining({ + id: published.id, + state: 'attempted', + }), + receipt: expect.objectContaining({ + channel: 'next-event', + invocationId: 'delivery-process', + }), + })]); + + const driver = createSqliteStateDriver({ file }); + const store = await driver.open(agentNoticeStateDefinition()); + const ledger = createAgentNoticeLedger(store, { + authorize: () => ({ state: 'authorized' }), + }); + const durable = await ledger.read(); + expect(durable.notices).toEqual([expect.objectContaining({ + id: published.id, + state: 'attempted', + attempts: [expect.objectContaining({ + invocationId: 'delivery-process', + })], + })]); + await driver.close(); + } finally { + await rm(root, { force: true, recursive: true }); + } + }); +}); diff --git a/packages/rsc-runtime/tests/state-packaging.test.ts b/packages/rsc-runtime/tests/state-packaging.test.ts index 68b42cba4..e178e57d8 100644 --- a/packages/rsc-runtime/tests/state-packaging.test.ts +++ b/packages/rsc-runtime/tests/state-packaging.test.ts @@ -20,7 +20,15 @@ const distFile = async (...segments: string[]): Promise => describe.sequential('state kernel packaging boundaries', () => { it('keeps every kernel and storage identifier out of the root and plugin entries', async () => { - const kernel = ['node:sqlite', 'defineState', 'AgentStateError', 'DatabaseSync', 'agent_state_journal'] as const; + const kernel = [ + 'node:sqlite', + 'defineState', + 'AgentStateError', + 'DatabaseSync', + 'agent_state_journal', + 'createAgentNoticeLedger', + 'agent-notice-ledger/v1', + ] as const; for (const entry of ['index.js', 'plugin.js']) { const source = await distFile(entry); for (const identifier of kernel) { @@ -42,6 +50,14 @@ describe.sequential('state kernel packaging boundaries', () => { expect(source).not.toContain('DatabaseSync'); }); + it('keeps the optional notice ledger out of stateless entries and node:sqlite out of its core entry', async () => { + const source = await distFile('notices.js'); + expect(source).toContain('createAgentNoticeLedger'); + expect(source).toContain('agent-notice-ledger/v1'); + expect(source).not.toContain('node:sqlite'); + expect(source).not.toContain('DatabaseSync'); + }); + it('gives the sqlite entry its own subpath that shares the state runtime', async () => { const source = await distFile('state', 'sqlite.js'); expect(source).toContain('node:sqlite'); @@ -49,7 +65,14 @@ describe.sequential('state kernel packaging boundaries', () => { const packageJson = JSON.parse(await readFile(join(packageRoot, 'package.json'), 'utf8')) as { exports: Record; }; - expect(Object.keys(packageJson.exports)).toEqual(['.', './plugin', './flight/server', './state', './state/sqlite']); + expect(Object.keys(packageJson.exports)).toEqual([ + '.', + './plugin', + './flight/server', + './state', + './state/sqlite', + './notices', + ]); for (const subpath of Object.keys(packageJson.exports)) { const target = packageJson.exports[subpath]!; await expect(distFile(...target.import.replace('./dist/', '').split('/'))).resolves.toBeTruthy(); diff --git a/rstest.integration-tests.ts b/rstest.integration-tests.ts index 105809059..b096c2494 100644 --- a/rstest.integration-tests.ts +++ b/rstest.integration-tests.ts @@ -46,6 +46,7 @@ export const integrationTestFiles: readonly string[] = [ 'packages/agent-bundle/tests/script-playground-service.test.ts', 'packages/agent-bundle/tests/target-hook-contract.test.ts', 'packages/agent-bundle/tests/target-mcp-runtime.test.ts', + 'packages/rsc-runtime/tests/notices-sqlite-cross-process.test.ts', 'packages/rsc-runtime/tests/state-packaging.test.ts', 'packages/rsc-runtime/tests/state-sqlite-cross-process.test.ts', 'packages/workbench/tests/comparisons-page-client-scope-browser.test.ts',