diff --git a/.changeset/conversational-control.md b/.changeset/conversational-control.md new file mode 100644 index 0000000..a0ab2a5 --- /dev/null +++ b/.changeset/conversational-control.md @@ -0,0 +1,7 @@ +--- +'@opencoven/cave-client': minor +--- + +Add the first bounded conversational-control authority to `@opencoven/cave-client` with Cave remaining the sole executor and canonical owner: canonical conversation create, idempotent send with one caller-visible operation UUID, explicit retry through a fresh operation UUID and `retryOfTurnId`, non-content operation reads, explicit stop, a typed resumable event stream driven by one translator for initial and resumed pages, operation-ID error propagation, no automatic replay after ambiguous transport completion, and `reconcile_required` helpers that instruct a canonical history reload. + +The five Client v1 conversation operations (`conversations.create`, `messages.send`, `operations.read`, `operations.events`, `operations.stop`) are not yet declared by the authoritative Cave contract fixture (pinned producer commit `4adc97b1`), so the optional `CaveTransport` bindings stay unbound and every call reports `unsupported_operation` until the upstream Cave mutation contract lands; no speculative routes ship. The private CLI streaming renderers follow in a separate PR per the design's PR plan. diff --git a/api-baselines/cave.d.ts b/api-baselines/cave.d.ts index 2bd9d70..f51b10f 100644 --- a/api-baselines/cave.d.ts +++ b/api-baselines/cave.d.ts @@ -1,6 +1,6 @@ // Entrypoint: . -// Declaration: dist/client-BbxpTVKf.d.ts -import { OperationContext, PageOptions, OperationDefaults, SecretStore, SecretStoreReference, OperationOptions, Page, BoundedPageOptions, NormalizedError, CompatibilityAssessment } from '@opencoven/sdk-core/browser'; +// Declaration: dist/client-ootQTXcj.d.ts +import { OperationObserver, OperationContext, PageOptions, OperationDefaults, SecretStore, SecretStoreReference, OperationOptions, Page, BoundedPageOptions, NormalizedError, CompatibilityAssessment } from '@opencoven/sdk-core/browser'; interface CaveCanonicalFamiliar { id: string; @@ -358,6 +358,186 @@ interface CaveFamiliarAnalyticsResponse { error?: string; } +/** + * Conversational control: the first bounded mutation authority. + * + * Cave remains the sole executor and canonical state owner. The SDK exposes + * constrained typed operations only — never arbitrary HTTP paths, private + * Cave routes, or raw transport escape hatches — and owns the public DTOs, + * validators, and the single event translator shared by initial and resumed + * streams. + * + * The five Client v1 operations this surface is defined against + * (`conversations.create`, `messages.send`, `operations.read`, + * `operations.events`, `operations.stop`) are not yet declared by the + * authoritative Cave contract fixture this SDK vendors. This module therefore + * defines the typed requests, results, operation records, event vocabulary, + * cursor handling, and translation rules only; it introduces no HTTP paths. + * Transport bindings for the five operations stay optional and are expected + * to arrive with the upstream Cave producer contract and a re-imported + * fixture. + * + * This module is import-pure: no discovery, credential, filesystem, network, + * or daemon I/O happens at import time. + */ +type CaveConversationOperationId = string; +type CaveConversationEventCursor = string; +interface CaveCreateConversationRequest { + operationId: CaveConversationOperationId; + familiarId: string; + projectId?: string; +} +type CaveSendConversationMessageRequest = { + operationId: CaveConversationOperationId; + text: string; + retryOfTurnId?: never; +} | { + operationId: CaveConversationOperationId; + retryOfTurnId: string; + text?: never; +}; +interface CaveRetryConversationTurnRequest { + operationId: CaveConversationOperationId; + retryOfTurnId: string; +} +type CaveConversationOperationState = 'pending' | 'accepted' | 'running' | 'stopping' | 'completed' | 'failed' | 'cancelled'; +type CaveConversationOperationKind = 'conversations.create' | 'messages.send'; +type CaveConversationOriginatingScope = 'chat:write' | 'conversations:write'; +interface CaveConversationOperation { + id: CaveConversationOperationId; + kind: CaveConversationOperationKind; + state: CaveConversationOperationState; + originatingScope: CaveConversationOriginatingScope; + conversationId: string; + inputTurnId?: string; + outputTurnId?: string; + retryOfTurnId?: string; + failureCode?: string; + latestEventId: number; + replayFloorEventId: number; + createdAt: string; + updatedAt: string; + idempotencyResultExpiresAt?: string; +} +interface CaveCreateConversationResult { + operationId: CaveConversationOperationId; + replayed: boolean; + conversation: CaveConversation; +} +interface CaveSendConversationMessageResult { + operation: CaveConversationOperation; + replayed: boolean; +} +interface CaveConversationEventBase { + operationId: CaveConversationOperationId; + eventId: number; + cursor: CaveConversationEventCursor; + occurredAt: string; +} +type CaveConversationEventType = 'operation.accepted' | 'assistant.delta' | 'operation.stopping' | 'operation.completed' | 'operation.failed' | 'operation.cancelled'; +type CaveConversationEvent = (CaveConversationEventBase & { + type: 'operation.accepted'; + conversationId: string; + inputTurnId: string; + retryOfTurnId?: string; +}) | (CaveConversationEventBase & { + type: 'assistant.delta'; + text: string; +}) | (CaveConversationEventBase & { + type: 'operation.stopping'; +}) | (CaveConversationEventBase & { + type: 'operation.completed'; + outputTurnId: string; +}) | (CaveConversationEventBase & { + type: 'operation.failed'; + outputTurnId: string; + code: string; +}) | (CaveConversationEventBase & { + type: 'operation.cancelled'; + outputTurnId: string; +}); +interface CaveConversationEventPage { + operation: CaveConversationOperation; + events: readonly CaveConversationEvent[]; + complete: boolean; + cursor?: { + current?: CaveConversationEventCursor; + next?: CaveConversationEventCursor; + hasMore: boolean; + }; +} +interface CaveConversationEventPageRequest { + cursor?: CaveConversationEventCursor; + waitMs?: number; +} +interface CaveConversationStreamOptions { + cursor?: CaveConversationEventCursor; + signal?: AbortSignal; + timeoutMs?: number; + observer?: OperationObserver; +} +declare const CAVE_CONVERSATION_OPERATION_STATES: readonly ["pending", "accepted", "running", "stopping", "completed", "failed", "cancelled"]; +declare const CAVE_CONVERSATION_TERMINAL_STATES: readonly ["completed", "failed", "cancelled"]; +declare const CAVE_CONVERSATION_EVENT_TYPES: readonly ["operation.accepted", "assistant.delta", "operation.stopping", "operation.completed", "operation.failed", "operation.cancelled"]; +/** The scope stored with an operation when it was claimed; reads of the operation and its events are authorized by it. */ +declare const CAVE_CONVERSATION_ORIGINATING_SCOPES: readonly ["chat:write", "conversations:write"]; +/** + * The defined `reconcile_required` reasons. A `reconcile_required` error is + * an instruction to reload canonical state, not a transient transport retry. + */ +declare const CAVE_CONVERSATION_RECONCILE_REASONS: readonly ["replay_gap", "operation_expired", "canonical_branch_changed", "idempotency_result_expired", "canonical_state_moved"]; +type CaveConversationReconcileReason = (typeof CAVE_CONVERSATION_RECONCILE_REASONS)[number]; +/** + * Event cursors are opaque route strings bounded by the authoritative + * `cursorCharacters` limit. The SDK never decodes them. + */ +declare function validateConversationEventCursor(value: unknown, label: string): CaveConversationEventCursor; +interface CaveConversationTranslatedPage { + operation: CaveConversationOperation; + events: readonly CaveConversationEvent[]; + complete: boolean; + requestId: string | undefined; + nextCursor?: CaveConversationEventCursor; +} +interface CaveConversationEventTranslator { + readonly operationId: CaveConversationOperationId; + readonly deliveredThroughEventId: number; + /** + * Validate one raw event-page response and return the accepted events in + * wire order. Throws a protocol error on any violation. + */ + translate(page: unknown): CaveConversationTranslatedPage; + /** Advance the accepted cursor after the event has been delivered. */ + commit(eventId: number): void; +} +interface CaveConversationEventTranslatorOptions { + /** + * True when this stream's first page resumes behind an opaque cursor from + * an earlier generator run: the first accepted event then cannot be + * gap-checked against a known event ID. A fresh stream must begin at + * event 1. + */ + resumeAfterOpaqueCursor?: boolean; +} +/** + * The one parser/translator for conversation event pages. Initial attachment + * and every resumed long poll pass through it. + * + * The translator validates the shared Client v1 envelope before event data, + * validates the operation ID on every event, requires contiguous increasing + * event IDs, suppresses an exact duplicate event at or below the caller's + * accepted cursor, and refuses forward gaps, reordered events, changed + * operation IDs, malformed terminal sequences, and events after terminal as + * protocol violations. The resume cursor advances only when the caller + * commits, after the corresponding event has been delivered. + */ +declare function createConversationEventTranslator(operationId: CaveConversationOperationId, options?: CaveConversationEventTranslatorOptions): CaveConversationEventTranslator; +/** + * The defined `reconcile_required` reasons, read from normalized error + * details without trusting the error shape. + */ +declare function caveConversationReconcileReason(error: unknown): CaveConversationReconcileReason | undefined; + interface CaveTransport { health(context?: OperationContext): Promise; pairingCreate?(request: CavePairingRequest, context?: OperationContext): Promise; @@ -372,6 +552,19 @@ interface CaveTransport { listConversations?(options: PageOptions, context?: OperationContext): Promise; getConversation?(conversationId: string, context?: OperationContext): Promise; listConversationMessages?(conversationId: string, options: PageOptions, context?: OperationContext): Promise; + /** + * Conversational control is optional for every transport. The five Client + * v1 conversation-operation routes are not yet declared by the + * authoritative Cave contract fixture this SDK vendors, so no transport + * binds them today; the client reports a missing one as + * `unsupported_operation` rather than inventing a route. Results are + * `unknown` at this trust boundary and are validated by the client. + */ + createConversation?(request: CaveCreateConversationRequest, context?: OperationContext): Promise; + sendConversationMessage?(conversationId: string, request: CaveSendConversationMessageRequest, context?: OperationContext): Promise; + getConversationOperation?(operationId: CaveConversationOperationId, context?: OperationContext): Promise; + readConversationOperationEvents?(operationId: CaveConversationOperationId, page: CaveConversationEventPageRequest, context?: OperationContext): Promise; + stopConversationOperation?(operationId: CaveConversationOperationId, context?: OperationContext): Promise; /** * The familiar operations are optional so that a transport written against * an older Cave still satisfies this interface. The client reports a missing @@ -453,6 +646,13 @@ declare class CaveClientError extends Error { readonly statusCode: number | undefined; readonly details: Record | undefined; constructor(normalized: NormalizedError, compatibility?: CompatibilityAssessment, options?: ErrorOptions); + /** + * The caller-visible operation UUID for a conversation mutation or stream, + * attached once the validated operation ID has been accepted by the SDK. + * Undefined for errors raised before acceptance and for non-conversation + * operations. Carries fixed metadata only. + */ + get operationId(): string | undefined; } declare function isCaveClientError(error: unknown): error is CaveClientError; declare class CavePairingSession { @@ -490,14 +690,59 @@ declare class CaveClient { createPairing(request: CavePairingRequest, options?: OperationOptions): Promise; credentialStatus(options?: OperationOptions): Promise; forgetCredential(options?: OperationOptions): Promise; + /** + * Canonical conversation creation. Accepts only the operation UUID, one + * canonical familiar ID, and an optional canonical project ID; Cave owns + * every other decision. Create does not start an executor and does not + * open an event stream. + */ + createConversation(request: CaveCreateConversationRequest, options?: OperationOptions): Promise; + /** + * One text send, or one explicit retry when the request carries + * `retryOfTurnId`. The response is an acceptance/result envelope, not the + * output stream; attach with `streamConversationOperation`. The exact text + * is preserved byte for byte. An identical completed send replays Cave's + * recorded result; the SDK never replays one on its own. + */ + sendConversationMessage(conversationId: string, request: CaveSendConversationMessageRequest, options?: OperationOptions): Promise; + /** + * Typed convenience over `messages.send` for retrying an explicitly failed + * or cancelled assistant turn. It uses a fresh operation UUID and the + * explicit `retryOfTurnId`; it introduces no second producer route. + */ + retryConversationTurn(conversationId: string, request: CaveRetryConversationTurnRequest, options?: OperationOptions): Promise; + /** + * The non-content operation record: fixed codes, turn references, event + * bounds, and timestamps. Prompt, attachment, bearer, and raw-cause + * content never appear here. + */ + getConversationOperation(operationId: string, options?: OperationOptions): Promise; + /** + * The typed, resumable event stream for one conversation operation. + * + * `options.timeoutMs` is one total stream budget; each long poll receives + * only the remaining budget. A caller abort closes the current event read + * and this generator: it never calls Stop and never resubmits a send. + * Initial attachment and every resumed page pass through the same event + * translator, which suppresses duplicates at or below the accepted cursor + * and refuses protocol violations. `reconcile_required` is an instruction + * to reload canonical history, not a retry. + */ + streamConversationOperation(operationId: string, options?: CaveConversationStreamOptions): AsyncGenerator; + /** + * Explicit Stop for one conversation operation. Repeated Stop calls are + * safe against the target operation; this client sends each Stop exactly + * once and never retries it after ambiguous transport completion. + */ + stopConversationOperation(operationId: string, options?: OperationOptions): Promise; } declare function createCaveClient(options: CaveClientOptions): CaveClient; -export { type CavePropertyCoverage as $, type CaveExecutionWindow as A, type CaveFamiliar as B, type CavePairingRequest as C, type CaveFamiliarAnalytics as D, type CaveFamiliarAnalyticsOptions as E, type CaveFamiliarAnalyticsResponse as F, type CaveFamiliarContract as G, type CaveFamiliarContractResponse as H, type CaveFamiliarProperty as I, type CaveFamiliarWire as J, type CaveFamiliarsResponse as K, type CaveHealth as L, type CaveHealthData as M, type CaveHealthResponse as N, type CaveManagedCredentialStatusResult as O, type CaveManagedCredentialTransport as P, type CaveManagedForgetCredentialResult as Q, type CaveManagedNativeCredentialCustody as R, type CaveManagedPairingCreated as S, type CaveManagedPairingExchange as T, type CavePairingCreated as U, type CavePairingExchange as V, type CavePairingScope as W, CavePairingSession as X, type CavePairingState as Y, type CavePairingStatus as Z, type CaveProject as _, CaveClient as a, type CaveTransport as a0, createCaveClient as a1, isCaveClientError as a2, normalizeCaveError as a3, CAVE_ANALYTICS_WINDOWS as b, CAVE_FAMILIAR_PROPERTIES as c, CAVE_PAIRING_SCOPES as d, CAVE_PAIRING_STATUSES as e, type CaveAnalyticsWindowKey as f, type CaveAuthorityBinding as g, type CaveAuthorityBoundPairingExchange as h, type CaveCanonicalFamiliar as i, CaveClientError as j, type CaveClientOptions as k, type CaveContractFile as l, type CaveContractReport as m, type CaveContractViolation as n, type CaveConversation as o, type CaveConversationMessage as p, type CaveCredentialAccess as q, type CaveCredentialBinding as r, type CaveCredentialDisconnectedReason as s, type CaveCredentialMetadata as t, type CaveCredentialPersistingTransport as u, type CaveCredentialStatus as v, type CaveExecutionAttempt as w, type CaveExecutionBackfill as x, type CaveExecutionCoverage as y, type CaveExecutionSlice as z }; +export { type CaveFamiliarContract as $, type CaveConversationMessage as A, type CaveConversationOperation as B, type CavePairingRequest as C, type CaveConversationOperationId as D, type CaveConversationOperationKind as E, type CaveConversationOperationState as F, type CaveConversationOriginatingScope as G, type CaveConversationReconcileReason as H, type CaveConversationStreamOptions as I, type CaveConversationTranslatedPage as J, type CaveCreateConversationRequest as K, type CaveCreateConversationResult as L, type CaveCredentialAccess as M, type CaveCredentialBinding as N, type CaveCredentialDisconnectedReason as O, type CaveCredentialMetadata as P, type CaveCredentialPersistingTransport as Q, type CaveCredentialStatus as R, type CaveExecutionAttempt as S, type CaveExecutionBackfill as T, type CaveExecutionCoverage as U, type CaveExecutionSlice as V, type CaveExecutionWindow as W, type CaveFamiliar as X, type CaveFamiliarAnalytics as Y, type CaveFamiliarAnalyticsOptions as Z, type CaveFamiliarAnalyticsResponse as _, CaveClient as a, type CaveFamiliarContractResponse as a0, type CaveFamiliarProperty as a1, type CaveFamiliarWire as a2, type CaveFamiliarsResponse as a3, type CaveHealth as a4, type CaveHealthData as a5, type CaveHealthResponse as a6, type CaveManagedCredentialStatusResult as a7, type CaveManagedCredentialTransport as a8, type CaveManagedForgetCredentialResult as a9, type CaveManagedNativeCredentialCustody as aa, type CaveManagedPairingCreated as ab, type CaveManagedPairingExchange as ac, type CavePairingCreated as ad, type CavePairingExchange as ae, type CavePairingScope as af, CavePairingSession as ag, type CavePairingState as ah, type CavePairingStatus as ai, type CaveProject as aj, type CavePropertyCoverage as ak, type CaveRetryConversationTurnRequest as al, type CaveSendConversationMessageRequest as am, type CaveSendConversationMessageResult as an, type CaveTransport as ao, caveConversationReconcileReason as ap, createCaveClient as aq, createConversationEventTranslator as ar, isCaveClientError as as, normalizeCaveError as at, validateConversationEventCursor as au, CAVE_ANALYTICS_WINDOWS as b, CAVE_CONVERSATION_EVENT_TYPES as c, CAVE_CONVERSATION_OPERATION_STATES as d, CAVE_CONVERSATION_ORIGINATING_SCOPES as e, CAVE_CONVERSATION_RECONCILE_REASONS as f, CAVE_CONVERSATION_TERMINAL_STATES as g, CAVE_FAMILIAR_PROPERTIES as h, CAVE_PAIRING_SCOPES as i, CAVE_PAIRING_STATUSES as j, type CaveAnalyticsWindowKey as k, type CaveAuthorityBinding as l, type CaveAuthorityBoundPairingExchange as m, type CaveCanonicalFamiliar as n, CaveClientError as o, type CaveClientOptions as p, type CaveContractFile as q, type CaveContractReport as r, type CaveContractViolation as s, type CaveConversation as t, type CaveConversationEvent as u, type CaveConversationEventBase as v, type CaveConversationEventPage as w, type CaveConversationEventPageRequest as x, type CaveConversationEventTranslator as y, type CaveConversationEventType as z }; // Entrypoint: . // Declaration: dist/index.d.ts -import { C as CavePairingRequest, a as CaveClient } from './client-BbxpTVKf.js'; -export { b as CAVE_ANALYTICS_WINDOWS, c as CAVE_FAMILIAR_PROPERTIES, d as CAVE_PAIRING_SCOPES, e as CAVE_PAIRING_STATUSES, f as CaveAnalyticsWindowKey, g as CaveAuthorityBinding, h as CaveAuthorityBoundPairingExchange, i as CaveCanonicalFamiliar, j as CaveClientError, k as CaveClientOptions, l as CaveContractFile, m as CaveContractReport, n as CaveContractViolation, o as CaveConversation, p as CaveConversationMessage, q as CaveCredentialAccess, r as CaveCredentialBinding, s as CaveCredentialDisconnectedReason, t as CaveCredentialMetadata, u as CaveCredentialPersistingTransport, v as CaveCredentialStatus, w as CaveExecutionAttempt, x as CaveExecutionBackfill, y as CaveExecutionCoverage, z as CaveExecutionSlice, A as CaveExecutionWindow, B as CaveFamiliar, D as CaveFamiliarAnalytics, E as CaveFamiliarAnalyticsOptions, F as CaveFamiliarAnalyticsResponse, G as CaveFamiliarContract, H as CaveFamiliarContractResponse, I as CaveFamiliarProperty, J as CaveFamiliarWire, K as CaveFamiliarsResponse, L as CaveHealth, M as CaveHealthData, N as CaveHealthResponse, O as CaveManagedCredentialStatusResult, P as CaveManagedCredentialTransport, Q as CaveManagedForgetCredentialResult, R as CaveManagedNativeCredentialCustody, S as CaveManagedPairingCreated, T as CaveManagedPairingExchange, U as CavePairingCreated, V as CavePairingExchange, W as CavePairingScope, X as CavePairingSession, Y as CavePairingState, Z as CavePairingStatus, _ as CaveProject, $ as CavePropertyCoverage, a0 as CaveTransport, a1 as createCaveClient, a2 as isCaveClientError, a3 as normalizeCaveError } from './client-BbxpTVKf.js'; +import { C as CavePairingRequest, a as CaveClient } from './client-ootQTXcj.js'; +export { b as CAVE_ANALYTICS_WINDOWS, c as CAVE_CONVERSATION_EVENT_TYPES, d as CAVE_CONVERSATION_OPERATION_STATES, e as CAVE_CONVERSATION_ORIGINATING_SCOPES, f as CAVE_CONVERSATION_RECONCILE_REASONS, g as CAVE_CONVERSATION_TERMINAL_STATES, h as CAVE_FAMILIAR_PROPERTIES, i as CAVE_PAIRING_SCOPES, j as CAVE_PAIRING_STATUSES, k as CaveAnalyticsWindowKey, l as CaveAuthorityBinding, m as CaveAuthorityBoundPairingExchange, n as CaveCanonicalFamiliar, o as CaveClientError, p as CaveClientOptions, q as CaveContractFile, r as CaveContractReport, s as CaveContractViolation, t as CaveConversation, u as CaveConversationEvent, v as CaveConversationEventBase, w as CaveConversationEventPage, x as CaveConversationEventPageRequest, y as CaveConversationEventTranslator, z as CaveConversationEventType, A as CaveConversationMessage, B as CaveConversationOperation, D as CaveConversationOperationId, E as CaveConversationOperationKind, F as CaveConversationOperationState, G as CaveConversationOriginatingScope, H as CaveConversationReconcileReason, I as CaveConversationStreamOptions, J as CaveConversationTranslatedPage, K as CaveCreateConversationRequest, L as CaveCreateConversationResult, M as CaveCredentialAccess, N as CaveCredentialBinding, O as CaveCredentialDisconnectedReason, P as CaveCredentialMetadata, Q as CaveCredentialPersistingTransport, R as CaveCredentialStatus, S as CaveExecutionAttempt, T as CaveExecutionBackfill, U as CaveExecutionCoverage, V as CaveExecutionSlice, W as CaveExecutionWindow, X as CaveFamiliar, Y as CaveFamiliarAnalytics, Z as CaveFamiliarAnalyticsOptions, _ as CaveFamiliarAnalyticsResponse, $ as CaveFamiliarContract, a0 as CaveFamiliarContractResponse, a1 as CaveFamiliarProperty, a2 as CaveFamiliarWire, a3 as CaveFamiliarsResponse, a4 as CaveHealth, a5 as CaveHealthData, a6 as CaveHealthResponse, a7 as CaveManagedCredentialStatusResult, a8 as CaveManagedCredentialTransport, a9 as CaveManagedForgetCredentialResult, aa as CaveManagedNativeCredentialCustody, ab as CaveManagedPairingCreated, ac as CaveManagedPairingExchange, ad as CavePairingCreated, ae as CavePairingExchange, af as CavePairingScope, ag as CavePairingSession, ah as CavePairingState, ai as CavePairingStatus, aj as CaveProject, ak as CavePropertyCoverage, al as CaveRetryConversationTurnRequest, am as CaveSendConversationMessageRequest, an as CaveSendConversationMessageResult, ao as CaveTransport, ap as caveConversationReconcileReason, aq as createCaveClient, ar as createConversationEventTranslator, as as isCaveClientError, at as normalizeCaveError, au as validateConversationEventCursor } from './client-ootQTXcj.js'; import { OperationOptions, OperationContext, PageOptions, OperationDefaults, SecretStore, SecretStoreReference } from '@opencoven/sdk-core'; import '@opencoven/sdk-core/browser'; @@ -794,8 +1039,8 @@ declare const CAVE_CLIENT_VERSION: string; export { CAVE_CLIENT_VERSION, CaveClient, type CaveContractCursor, type CaveContractEnvelopeMetadata, type CaveContractFixture, type CaveContractHealthData, type CaveContractIdentity, type CaveContractOperation, type CaveContractPairingCreatedData, type CaveContractPairingExchangeData, type CaveContractPairingStatusData, type CaveContractPublicRoute, type CaveContractRevision, type CaveDiscoveredClientOptions, type CaveDiscoveredEndpoint, type CaveDiscoveryDependencies, CaveDiscoveryError, type CaveDiscoveryErrorCode, type CaveDiscoveryFileHandle, type CaveDiscoveryPathIdentity, type CaveDiscoveryRecordIdentity, type CaveEndpointFreshness, type CaveManagedClientOptions, type CaveManagedNativeDiscardResult, type CaveManagedNativePairingCreated, type CaveManagedNativePairingExchange, type CaveManagedNativeResponse, type CaveManagedNativeTransport, CavePairingRequest, type CaveWindowsPathTrustResult, type CaveWindowsPathTrustValidator, type DiscoverCaveEndpointOptions, createDiscoveredCaveClient, createManagedCaveClient, digestCaveContractFixture, discoverCaveEndpoint, isCaveDiscoveryError, parseCaveContractFixture, parseVerifiedCaveContractFixture, verifyCaveContractFixtureDigest }; // Entrypoint: ./managed -// Declaration: dist/client-BbxpTVKf.d.ts -import { OperationContext, PageOptions, OperationDefaults, SecretStore, SecretStoreReference, OperationOptions, Page, BoundedPageOptions, NormalizedError, CompatibilityAssessment } from '@opencoven/sdk-core/browser'; +// Declaration: dist/client-ootQTXcj.d.ts +import { OperationObserver, OperationContext, PageOptions, OperationDefaults, SecretStore, SecretStoreReference, OperationOptions, Page, BoundedPageOptions, NormalizedError, CompatibilityAssessment } from '@opencoven/sdk-core/browser'; interface CaveCanonicalFamiliar { id: string; @@ -1153,6 +1398,186 @@ interface CaveFamiliarAnalyticsResponse { error?: string; } +/** + * Conversational control: the first bounded mutation authority. + * + * Cave remains the sole executor and canonical state owner. The SDK exposes + * constrained typed operations only — never arbitrary HTTP paths, private + * Cave routes, or raw transport escape hatches — and owns the public DTOs, + * validators, and the single event translator shared by initial and resumed + * streams. + * + * The five Client v1 operations this surface is defined against + * (`conversations.create`, `messages.send`, `operations.read`, + * `operations.events`, `operations.stop`) are not yet declared by the + * authoritative Cave contract fixture this SDK vendors. This module therefore + * defines the typed requests, results, operation records, event vocabulary, + * cursor handling, and translation rules only; it introduces no HTTP paths. + * Transport bindings for the five operations stay optional and are expected + * to arrive with the upstream Cave producer contract and a re-imported + * fixture. + * + * This module is import-pure: no discovery, credential, filesystem, network, + * or daemon I/O happens at import time. + */ +type CaveConversationOperationId = string; +type CaveConversationEventCursor = string; +interface CaveCreateConversationRequest { + operationId: CaveConversationOperationId; + familiarId: string; + projectId?: string; +} +type CaveSendConversationMessageRequest = { + operationId: CaveConversationOperationId; + text: string; + retryOfTurnId?: never; +} | { + operationId: CaveConversationOperationId; + retryOfTurnId: string; + text?: never; +}; +interface CaveRetryConversationTurnRequest { + operationId: CaveConversationOperationId; + retryOfTurnId: string; +} +type CaveConversationOperationState = 'pending' | 'accepted' | 'running' | 'stopping' | 'completed' | 'failed' | 'cancelled'; +type CaveConversationOperationKind = 'conversations.create' | 'messages.send'; +type CaveConversationOriginatingScope = 'chat:write' | 'conversations:write'; +interface CaveConversationOperation { + id: CaveConversationOperationId; + kind: CaveConversationOperationKind; + state: CaveConversationOperationState; + originatingScope: CaveConversationOriginatingScope; + conversationId: string; + inputTurnId?: string; + outputTurnId?: string; + retryOfTurnId?: string; + failureCode?: string; + latestEventId: number; + replayFloorEventId: number; + createdAt: string; + updatedAt: string; + idempotencyResultExpiresAt?: string; +} +interface CaveCreateConversationResult { + operationId: CaveConversationOperationId; + replayed: boolean; + conversation: CaveConversation; +} +interface CaveSendConversationMessageResult { + operation: CaveConversationOperation; + replayed: boolean; +} +interface CaveConversationEventBase { + operationId: CaveConversationOperationId; + eventId: number; + cursor: CaveConversationEventCursor; + occurredAt: string; +} +type CaveConversationEventType = 'operation.accepted' | 'assistant.delta' | 'operation.stopping' | 'operation.completed' | 'operation.failed' | 'operation.cancelled'; +type CaveConversationEvent = (CaveConversationEventBase & { + type: 'operation.accepted'; + conversationId: string; + inputTurnId: string; + retryOfTurnId?: string; +}) | (CaveConversationEventBase & { + type: 'assistant.delta'; + text: string; +}) | (CaveConversationEventBase & { + type: 'operation.stopping'; +}) | (CaveConversationEventBase & { + type: 'operation.completed'; + outputTurnId: string; +}) | (CaveConversationEventBase & { + type: 'operation.failed'; + outputTurnId: string; + code: string; +}) | (CaveConversationEventBase & { + type: 'operation.cancelled'; + outputTurnId: string; +}); +interface CaveConversationEventPage { + operation: CaveConversationOperation; + events: readonly CaveConversationEvent[]; + complete: boolean; + cursor?: { + current?: CaveConversationEventCursor; + next?: CaveConversationEventCursor; + hasMore: boolean; + }; +} +interface CaveConversationEventPageRequest { + cursor?: CaveConversationEventCursor; + waitMs?: number; +} +interface CaveConversationStreamOptions { + cursor?: CaveConversationEventCursor; + signal?: AbortSignal; + timeoutMs?: number; + observer?: OperationObserver; +} +declare const CAVE_CONVERSATION_OPERATION_STATES: readonly ["pending", "accepted", "running", "stopping", "completed", "failed", "cancelled"]; +declare const CAVE_CONVERSATION_TERMINAL_STATES: readonly ["completed", "failed", "cancelled"]; +declare const CAVE_CONVERSATION_EVENT_TYPES: readonly ["operation.accepted", "assistant.delta", "operation.stopping", "operation.completed", "operation.failed", "operation.cancelled"]; +/** The scope stored with an operation when it was claimed; reads of the operation and its events are authorized by it. */ +declare const CAVE_CONVERSATION_ORIGINATING_SCOPES: readonly ["chat:write", "conversations:write"]; +/** + * The defined `reconcile_required` reasons. A `reconcile_required` error is + * an instruction to reload canonical state, not a transient transport retry. + */ +declare const CAVE_CONVERSATION_RECONCILE_REASONS: readonly ["replay_gap", "operation_expired", "canonical_branch_changed", "idempotency_result_expired", "canonical_state_moved"]; +type CaveConversationReconcileReason = (typeof CAVE_CONVERSATION_RECONCILE_REASONS)[number]; +/** + * Event cursors are opaque route strings bounded by the authoritative + * `cursorCharacters` limit. The SDK never decodes them. + */ +declare function validateConversationEventCursor(value: unknown, label: string): CaveConversationEventCursor; +interface CaveConversationTranslatedPage { + operation: CaveConversationOperation; + events: readonly CaveConversationEvent[]; + complete: boolean; + requestId: string | undefined; + nextCursor?: CaveConversationEventCursor; +} +interface CaveConversationEventTranslator { + readonly operationId: CaveConversationOperationId; + readonly deliveredThroughEventId: number; + /** + * Validate one raw event-page response and return the accepted events in + * wire order. Throws a protocol error on any violation. + */ + translate(page: unknown): CaveConversationTranslatedPage; + /** Advance the accepted cursor after the event has been delivered. */ + commit(eventId: number): void; +} +interface CaveConversationEventTranslatorOptions { + /** + * True when this stream's first page resumes behind an opaque cursor from + * an earlier generator run: the first accepted event then cannot be + * gap-checked against a known event ID. A fresh stream must begin at + * event 1. + */ + resumeAfterOpaqueCursor?: boolean; +} +/** + * The one parser/translator for conversation event pages. Initial attachment + * and every resumed long poll pass through it. + * + * The translator validates the shared Client v1 envelope before event data, + * validates the operation ID on every event, requires contiguous increasing + * event IDs, suppresses an exact duplicate event at or below the caller's + * accepted cursor, and refuses forward gaps, reordered events, changed + * operation IDs, malformed terminal sequences, and events after terminal as + * protocol violations. The resume cursor advances only when the caller + * commits, after the corresponding event has been delivered. + */ +declare function createConversationEventTranslator(operationId: CaveConversationOperationId, options?: CaveConversationEventTranslatorOptions): CaveConversationEventTranslator; +/** + * The defined `reconcile_required` reasons, read from normalized error + * details without trusting the error shape. + */ +declare function caveConversationReconcileReason(error: unknown): CaveConversationReconcileReason | undefined; + interface CaveTransport { health(context?: OperationContext): Promise; pairingCreate?(request: CavePairingRequest, context?: OperationContext): Promise; @@ -1167,6 +1592,19 @@ interface CaveTransport { listConversations?(options: PageOptions, context?: OperationContext): Promise; getConversation?(conversationId: string, context?: OperationContext): Promise; listConversationMessages?(conversationId: string, options: PageOptions, context?: OperationContext): Promise; + /** + * Conversational control is optional for every transport. The five Client + * v1 conversation-operation routes are not yet declared by the + * authoritative Cave contract fixture this SDK vendors, so no transport + * binds them today; the client reports a missing one as + * `unsupported_operation` rather than inventing a route. Results are + * `unknown` at this trust boundary and are validated by the client. + */ + createConversation?(request: CaveCreateConversationRequest, context?: OperationContext): Promise; + sendConversationMessage?(conversationId: string, request: CaveSendConversationMessageRequest, context?: OperationContext): Promise; + getConversationOperation?(operationId: CaveConversationOperationId, context?: OperationContext): Promise; + readConversationOperationEvents?(operationId: CaveConversationOperationId, page: CaveConversationEventPageRequest, context?: OperationContext): Promise; + stopConversationOperation?(operationId: CaveConversationOperationId, context?: OperationContext): Promise; /** * The familiar operations are optional so that a transport written against * an older Cave still satisfies this interface. The client reports a missing @@ -1248,6 +1686,13 @@ declare class CaveClientError extends Error { readonly statusCode: number | undefined; readonly details: Record | undefined; constructor(normalized: NormalizedError, compatibility?: CompatibilityAssessment, options?: ErrorOptions); + /** + * The caller-visible operation UUID for a conversation mutation or stream, + * attached once the validated operation ID has been accepted by the SDK. + * Undefined for errors raised before acceptance and for non-conversation + * operations. Carries fixed metadata only. + */ + get operationId(): string | undefined; } declare function isCaveClientError(error: unknown): error is CaveClientError; declare class CavePairingSession { @@ -1285,14 +1730,59 @@ declare class CaveClient { createPairing(request: CavePairingRequest, options?: OperationOptions): Promise; credentialStatus(options?: OperationOptions): Promise; forgetCredential(options?: OperationOptions): Promise; + /** + * Canonical conversation creation. Accepts only the operation UUID, one + * canonical familiar ID, and an optional canonical project ID; Cave owns + * every other decision. Create does not start an executor and does not + * open an event stream. + */ + createConversation(request: CaveCreateConversationRequest, options?: OperationOptions): Promise; + /** + * One text send, or one explicit retry when the request carries + * `retryOfTurnId`. The response is an acceptance/result envelope, not the + * output stream; attach with `streamConversationOperation`. The exact text + * is preserved byte for byte. An identical completed send replays Cave's + * recorded result; the SDK never replays one on its own. + */ + sendConversationMessage(conversationId: string, request: CaveSendConversationMessageRequest, options?: OperationOptions): Promise; + /** + * Typed convenience over `messages.send` for retrying an explicitly failed + * or cancelled assistant turn. It uses a fresh operation UUID and the + * explicit `retryOfTurnId`; it introduces no second producer route. + */ + retryConversationTurn(conversationId: string, request: CaveRetryConversationTurnRequest, options?: OperationOptions): Promise; + /** + * The non-content operation record: fixed codes, turn references, event + * bounds, and timestamps. Prompt, attachment, bearer, and raw-cause + * content never appear here. + */ + getConversationOperation(operationId: string, options?: OperationOptions): Promise; + /** + * The typed, resumable event stream for one conversation operation. + * + * `options.timeoutMs` is one total stream budget; each long poll receives + * only the remaining budget. A caller abort closes the current event read + * and this generator: it never calls Stop and never resubmits a send. + * Initial attachment and every resumed page pass through the same event + * translator, which suppresses duplicates at or below the accepted cursor + * and refuses protocol violations. `reconcile_required` is an instruction + * to reload canonical history, not a retry. + */ + streamConversationOperation(operationId: string, options?: CaveConversationStreamOptions): AsyncGenerator; + /** + * Explicit Stop for one conversation operation. Repeated Stop calls are + * safe against the target operation; this client sends each Stop exactly + * once and never retries it after ambiguous transport completion. + */ + stopConversationOperation(operationId: string, options?: OperationOptions): Promise; } declare function createCaveClient(options: CaveClientOptions): CaveClient; -export { type CavePropertyCoverage as $, type CaveExecutionWindow as A, type CaveFamiliar as B, type CavePairingRequest as C, type CaveFamiliarAnalytics as D, type CaveFamiliarAnalyticsOptions as E, type CaveFamiliarAnalyticsResponse as F, type CaveFamiliarContract as G, type CaveFamiliarContractResponse as H, type CaveFamiliarProperty as I, type CaveFamiliarWire as J, type CaveFamiliarsResponse as K, type CaveHealth as L, type CaveHealthData as M, type CaveHealthResponse as N, type CaveManagedCredentialStatusResult as O, type CaveManagedCredentialTransport as P, type CaveManagedForgetCredentialResult as Q, type CaveManagedNativeCredentialCustody as R, type CaveManagedPairingCreated as S, type CaveManagedPairingExchange as T, type CavePairingCreated as U, type CavePairingExchange as V, type CavePairingScope as W, CavePairingSession as X, type CavePairingState as Y, type CavePairingStatus as Z, type CaveProject as _, CaveClient as a, type CaveTransport as a0, createCaveClient as a1, isCaveClientError as a2, normalizeCaveError as a3, CAVE_ANALYTICS_WINDOWS as b, CAVE_FAMILIAR_PROPERTIES as c, CAVE_PAIRING_SCOPES as d, CAVE_PAIRING_STATUSES as e, type CaveAnalyticsWindowKey as f, type CaveAuthorityBinding as g, type CaveAuthorityBoundPairingExchange as h, type CaveCanonicalFamiliar as i, CaveClientError as j, type CaveClientOptions as k, type CaveContractFile as l, type CaveContractReport as m, type CaveContractViolation as n, type CaveConversation as o, type CaveConversationMessage as p, type CaveCredentialAccess as q, type CaveCredentialBinding as r, type CaveCredentialDisconnectedReason as s, type CaveCredentialMetadata as t, type CaveCredentialPersistingTransport as u, type CaveCredentialStatus as v, type CaveExecutionAttempt as w, type CaveExecutionBackfill as x, type CaveExecutionCoverage as y, type CaveExecutionSlice as z }; +export { type CaveFamiliarContract as $, type CaveConversationMessage as A, type CaveConversationOperation as B, type CavePairingRequest as C, type CaveConversationOperationId as D, type CaveConversationOperationKind as E, type CaveConversationOperationState as F, type CaveConversationOriginatingScope as G, type CaveConversationReconcileReason as H, type CaveConversationStreamOptions as I, type CaveConversationTranslatedPage as J, type CaveCreateConversationRequest as K, type CaveCreateConversationResult as L, type CaveCredentialAccess as M, type CaveCredentialBinding as N, type CaveCredentialDisconnectedReason as O, type CaveCredentialMetadata as P, type CaveCredentialPersistingTransport as Q, type CaveCredentialStatus as R, type CaveExecutionAttempt as S, type CaveExecutionBackfill as T, type CaveExecutionCoverage as U, type CaveExecutionSlice as V, type CaveExecutionWindow as W, type CaveFamiliar as X, type CaveFamiliarAnalytics as Y, type CaveFamiliarAnalyticsOptions as Z, type CaveFamiliarAnalyticsResponse as _, CaveClient as a, type CaveFamiliarContractResponse as a0, type CaveFamiliarProperty as a1, type CaveFamiliarWire as a2, type CaveFamiliarsResponse as a3, type CaveHealth as a4, type CaveHealthData as a5, type CaveHealthResponse as a6, type CaveManagedCredentialStatusResult as a7, type CaveManagedCredentialTransport as a8, type CaveManagedForgetCredentialResult as a9, type CaveManagedNativeCredentialCustody as aa, type CaveManagedPairingCreated as ab, type CaveManagedPairingExchange as ac, type CavePairingCreated as ad, type CavePairingExchange as ae, type CavePairingScope as af, CavePairingSession as ag, type CavePairingState as ah, type CavePairingStatus as ai, type CaveProject as aj, type CavePropertyCoverage as ak, type CaveRetryConversationTurnRequest as al, type CaveSendConversationMessageRequest as am, type CaveSendConversationMessageResult as an, type CaveTransport as ao, caveConversationReconcileReason as ap, createCaveClient as aq, createConversationEventTranslator as ar, isCaveClientError as as, normalizeCaveError as at, validateConversationEventCursor as au, CAVE_ANALYTICS_WINDOWS as b, CAVE_CONVERSATION_EVENT_TYPES as c, CAVE_CONVERSATION_OPERATION_STATES as d, CAVE_CONVERSATION_ORIGINATING_SCOPES as e, CAVE_CONVERSATION_RECONCILE_REASONS as f, CAVE_CONVERSATION_TERMINAL_STATES as g, CAVE_FAMILIAR_PROPERTIES as h, CAVE_PAIRING_SCOPES as i, CAVE_PAIRING_STATUSES as j, type CaveAnalyticsWindowKey as k, type CaveAuthorityBinding as l, type CaveAuthorityBoundPairingExchange as m, type CaveCanonicalFamiliar as n, CaveClientError as o, type CaveClientOptions as p, type CaveContractFile as q, type CaveContractReport as r, type CaveContractViolation as s, type CaveConversation as t, type CaveConversationEvent as u, type CaveConversationEventBase as v, type CaveConversationEventPage as w, type CaveConversationEventPageRequest as x, type CaveConversationEventTranslator as y, type CaveConversationEventType as z }; // Entrypoint: ./managed // Declaration: dist/managed.d.ts -import { P as CaveManagedCredentialTransport, a as CaveClient } from './client-BbxpTVKf.js'; -export { b as CAVE_ANALYTICS_WINDOWS, c as CAVE_FAMILIAR_PROPERTIES, d as CAVE_PAIRING_SCOPES, e as CAVE_PAIRING_STATUSES, i as CaveCanonicalFamiliar, j as CaveClientError, k as CaveClientOptions, o as CaveConversation, p as CaveConversationMessage, q as CaveCredentialAccess, r as CaveCredentialBinding, t as CaveCredentialMetadata, v as CaveCredentialStatus, E as CaveFamiliarAnalyticsOptions, L as CaveHealth, O as CaveManagedCredentialStatusResult, Q as CaveManagedForgetCredentialResult, R as CaveManagedNativeCredentialCustody, S as CaveManagedPairingCreated, T as CaveManagedPairingExchange, C as CavePairingRequest, W as CavePairingScope, X as CavePairingSession, Y as CavePairingState, Z as CavePairingStatus, _ as CaveProject, a0 as CaveTransport, a2 as isCaveClientError, a3 as normalizeCaveError } from './client-BbxpTVKf.js'; +import { a8 as CaveManagedCredentialTransport, a as CaveClient } from './client-ootQTXcj.js'; +export { b as CAVE_ANALYTICS_WINDOWS, h as CAVE_FAMILIAR_PROPERTIES, i as CAVE_PAIRING_SCOPES, j as CAVE_PAIRING_STATUSES, n as CaveCanonicalFamiliar, o as CaveClientError, p as CaveClientOptions, t as CaveConversation, A as CaveConversationMessage, M as CaveCredentialAccess, N as CaveCredentialBinding, P as CaveCredentialMetadata, R as CaveCredentialStatus, Z as CaveFamiliarAnalyticsOptions, a4 as CaveHealth, a7 as CaveManagedCredentialStatusResult, a9 as CaveManagedForgetCredentialResult, aa as CaveManagedNativeCredentialCustody, ab as CaveManagedPairingCreated, ac as CaveManagedPairingExchange, C as CavePairingRequest, af as CavePairingScope, ag as CavePairingSession, ah as CavePairingState, ai as CavePairingStatus, aj as CaveProject, ao as CaveTransport, as as isCaveClientError, at as normalizeCaveError } from './client-ootQTXcj.js'; import { OperationOptions, OperationDefaults, OperationContext } from '@opencoven/sdk-core/browser'; interface CaveManagedDiscoverySource { diff --git a/api-baselines/cave.json b/api-baselines/cave.json index ed10e7e..7838096 100644 --- a/api-baselines/cave.json +++ b/api-baselines/cave.json @@ -17,13 +17,18 @@ "entrypoints": { ".": { "declarationFiles": [ - "dist/client-BbxpTVKf.d.ts", + "dist/client-ootQTXcj.d.ts", "dist/index.d.ts" ], "runtimeExports": { "dist/index.js": [ "CAVE_ANALYTICS_WINDOWS", "CAVE_CLIENT_VERSION", + "CAVE_CONVERSATION_EVENT_TYPES", + "CAVE_CONVERSATION_OPERATION_STATES", + "CAVE_CONVERSATION_ORIGINATING_SCOPES", + "CAVE_CONVERSATION_RECONCILE_REASONS", + "CAVE_CONVERSATION_TERMINAL_STATES", "CAVE_FAMILIAR_PROPERTIES", "CAVE_PAIRING_SCOPES", "CAVE_PAIRING_STATUSES", @@ -31,7 +36,9 @@ "CaveClientError", "CaveDiscoveryError", "CavePairingSession", + "caveConversationReconcileReason", "createCaveClient", + "createConversationEventTranslator", "createDiscoveredCaveClient", "createManagedCaveClient", "digestCaveContractFixture", @@ -41,13 +48,14 @@ "normalizeCaveError", "parseCaveContractFixture", "parseVerifiedCaveContractFixture", + "validateConversationEventCursor", "verifyCaveContractFixtureDigest" ] } }, "./managed": { "declarationFiles": [ - "dist/client-BbxpTVKf.d.ts", + "dist/client-ootQTXcj.d.ts", "dist/managed.d.ts" ], "runtimeExports": { diff --git a/packages/cave/README.md b/packages/cave/README.md index 9beea46..f0aeca6 100644 --- a/packages/cave/README.md +++ b/packages/cave/README.md @@ -55,6 +55,9 @@ record. Unix discovery still requires a positive inode. one-page Client v1 canonical reads through `listFamiliars()`, `listProjects()`, `listConversations()`, `getConversation()`, and `listConversationMessages()`. +- Conversational control adds the first bounded mutation authority while Cave + remains the sole executor and canonical owner; see + [Conversational control](#conversational-control) below. - Four bounded async iterators, `iterateFamiliars()`, `iterateProjects()`, `iterateConversations()`, and `iterateConversationMessages()`, lazily compose the list routes. There is intentionally no iterator for the single-item @@ -472,6 +475,74 @@ return the complete Client v1 health envelope shown above. Consumers may keep checking `health.status`, and can additionally gate pairing or feature use from the normalized metadata. +## Conversational control + +Conversational control is the first bounded mutation authority. Cave remains +the sole executor, idempotency authority, operation journal, replay authority, +stop authority, and canonical conversation owner; the SDK exposes only +constrained typed operations — never arbitrary HTTP paths, private Cave +routes, or raw transport escape hatches. + +### Client methods + +- `client.createConversation({ operationId, familiarId, projectId? }, options?)` + creates one empty canonical conversation for one familiar. The caller + supplies the operation UUID; Cave resolves roots, harnesses, runtimes, + titles, and origin internally. +- `client.sendConversationMessage(conversationId, { operationId, text })` sends + one text message and returns the acceptance envelope with the operation + record and a `replayed` flag. Text is preserved byte for byte. A retry is + the same route with `retryOfTurnId` instead of text: + `client.retryConversationTurn(conversationId, { operationId, retryOfTurnId })` + is a typed convenience that introduces no second producer route. +- `client.getConversationOperation(operationId)` returns the non-content + operation record (fixed codes, turn references, event bounds, timestamps — + never prompt, attachment, or bearer content). +- `client.streamConversationOperation(operationId, options?)` returns the + typed, resumable event stream. `options.timeoutMs` is one total stream + budget; each long poll receives only the remaining budget. A caller abort + closes the current event read and the generator: it never calls Stop and + never resubmits a send. +- `client.stopConversationOperation(operationId)` sends each explicit Stop + exactly once and never retries it after an ambiguous transport completion; + calling Stop again explicitly is safe. + +Every mutation is dispatched exactly once. An ambiguous transport completion +never causes an automatic replay: inspect `error.operationId` (attached to +every post-acceptance error) and decide explicitly. An identical completed +mutation replays Cave's recorded result with `replayed: true`; a reused +operation key with a different canonical request hash is Cave's +`conflict / idempotency_key_reused`. Initial attachment and every resumed +stream pass through the same event translator, which validates the envelope +before event data, requires contiguous monotonic event IDs, suppresses exact +duplicates at or below the accepted cursor, and refuses gaps, reordering, +foreign operation IDs, and malformed terminal sequences as +`invalid_response`. On `reconcile_required`, reload `getConversation()` and +`listConversationMessages()` from the first page and replace — never append +to — the local projection; the SDK never fabricates omitted deltas. + +### Upstream contract gap + +The five Client v1 conversation operations (`conversations.create`, +`messages.send`, `operations.read`, `operations.events`, `operations.stop`) +are **not yet declared** by the authoritative Cave contract fixture this SDK +vendors (pinned producer commit `4adc97b1`). The SDK therefore ships the full +typed surface — request validation, result/event DTO parsing, the single +event translator for initial and resumed streams, operation-ID error +propagation, no-auto-replay semantics, and reconciliation helpers — while the +optional `CaveTransport` methods `createConversation`, +`sendConversationMessage`, `getConversationOperation`, +`readConversationOperationEvents`, and `stopConversationOperation` stay +unbound: every transport binding would be a speculative route. Calls today +fail with `unsupported_operation` naming the missing capability. The route +records, generated-fixture limits, event/cursor contract, and +request-hash conformance vectors are owed by the upstream Cave producer +contract; once that lands, `pnpm sync:contracts` imports the exact fixture +commit and transport bindings can be reviewed against it. The private CLI +streaming renderers (human/JSON/NDJSON ordering) are likewise staged for a +follow-up PR per the design's PR plan, as no CLI command can execute a +mutation against a real authority before that contract exists. + ## Compatibility, deadlines, and retry guidance Cave Client v1 health accepts additive Cave API updates on major version `1` diff --git a/packages/cave/src/canonical-reads.ts b/packages/cave/src/canonical-reads.ts index d9034b2..7085842 100644 --- a/packages/cave/src/canonical-reads.ts +++ b/packages/cave/src/canonical-reads.ts @@ -478,7 +478,7 @@ function parseProject(value: unknown, field: string): CaveProject { }; } -function parseConversation( +export function parseConversation( value: unknown, field: string, ): CaveConversation { diff --git a/packages/cave/src/client.ts b/packages/cave/src/client.ts index a75bdbb..9952e4d 100644 --- a/packages/cave/src/client.ts +++ b/packages/cave/src/client.ts @@ -25,6 +25,30 @@ import { parseCaveAuthorityBinding, } from './authority-binding-contract.js'; import { CAVE_CONTRACT_ERROR_CODES } from './contract-constraints.js'; +import { + CaveConversationResponseError, + CaveConversationSchemaError, + createConversationEventTranslator, + parseConversationOperationResponse, + parseCreateConversationRequest, + parseCreateConversationResult, + parseRetryConversationTurnRequest, + parseSendConversationMessageRequest, + parseSendConversationMessageResult, + validateConversationEventCursor, + validateConversationOperationId, + type CaveConversationEvent, + type CaveConversationEventCursor, + type CaveConversationOperation, + type CaveConversationOperationId, + type CaveConversationStreamOptions, + type CaveConversationTranslatedPage, + type CaveCreateConversationRequest, + type CaveCreateConversationResult, + type CaveRetryConversationTurnRequest, + type CaveSendConversationMessageRequest, + type CaveSendConversationMessageResult, +} from './conversation-control.js'; import { CaveCanonicalSchemaError, parseConversationEnvelope, @@ -226,6 +250,34 @@ export class CaveClientError extends Error { this.details = asStringRecord(ownDataErrorShape(options?.cause).details); Object.defineProperty(this, CAVE_CLIENT_ERROR_BRAND, { value: true }); } + + /** + * The caller-visible operation UUID for a conversation mutation or stream, + * attached once the validated operation ID has been accepted by the SDK. + * Undefined for errors raised before acceptance and for non-conversation + * operations. Carries fixed metadata only. + */ + get operationId(): string | undefined { + return CAVE_CONVERSATION_OPERATION_IDS.get(this); + } +} + +/** + * Operation IDs travel with conversation mutation and stream errors once the + * validated ID was accepted. A WeakMap keeps the error surface additive: + * the identifier is readable via `operationId` without changing any + * constructor or serialized shape. + */ +const CAVE_CONVERSATION_OPERATION_IDS = new WeakMap(); + +function attachConversationOperationId( + error: TError, + operationId: CaveConversationOperationId, +): TError { + if (isCaveClientError(error)) { + CAVE_CONVERSATION_OPERATION_IDS.set(error, operationId); + } + return error; } export function isCaveClientError(error: unknown): error is CaveClientError { @@ -3246,6 +3298,379 @@ export class CaveClient { return await forgetStoredCredential(credentials.store, credentials.reference, { context }); }, true, this.#stagedManagedCredentialTransport !== undefined); } + + /** + * One-shot conversation mutation. The executor runs exactly once: an + * ambiguous transport completion never replays the mutation, and the + * caller-visible operation UUID rides every post-acceptance error. + */ + /** + * One-shot conversation mutation. The executor runs exactly once: an + * ambiguous transport completion never replays the mutation, and the + * caller-visible operation UUID rides every post-acceptance error. + */ + async #conversationMutation( + operation: string, + operationId: CaveConversationOperationId, + options: OperationOptions, + executor: (context: OperationContext) => Promise, + ): Promise { + const redactManagedErrors = this.#usesManagedCredentialTransport(); + try { + return await this.#executePersistentMutation( + operation, + options, + async (context) => { + this.#ensureActive(context, operation); + return await executor(context); + }, + redactManagedErrors, + ); + } catch (error) { + throw attachConversationOperationId(error, operationId); + } + } + + #parseConversationResponse(operation: string, parse: () => T): T { + try { + return parse(); + } catch (error) { + if ( + error instanceof CaveConversationSchemaError || + error instanceof CaveCanonicalSchemaError + ) { + throw invalidCanonicalResponse( + operation, + error instanceof CaveCanonicalSchemaError + ? error.field + : (error).field, + ); + } + throw error; + } + } + + async #conversationTransportCall( + operation: string, + call: (() => Promise) | undefined, + ): Promise { + if (call === undefined) { + throw unsupported(operation); + } + return await call(); + } + + /** + * Canonical conversation creation. Accepts only the operation UUID, one + * canonical familiar ID, and an optional canonical project ID; Cave owns + * every other decision. Create does not start an executor and does not + * open an event stream. + */ + async createConversation( + request: CaveCreateConversationRequest, + options: OperationOptions = {}, + ): Promise { + const validated = parseCreateConversationRequest(request); + + return await this.#conversationMutation( + 'createConversation', + validated.operationId, + options, + async (context) => { + const call = this.#transport.createConversation?.bind(this.#transport); + const response = this.#managedSnapshot( + await this.#conversationTransportCall( + 'createConversation', + call === undefined ? undefined : () => call(validated, context), + ), + 'createConversation', + ); + return this.#parseConversationResponse( + 'createConversation', + () => parseCreateConversationResult(response, validated.operationId), + ); + }, + ); + } + + /** + * One text send, or one explicit retry when the request carries + * `retryOfTurnId`. The response is an acceptance/result envelope, not the + * output stream; attach with `streamConversationOperation`. The exact text + * is preserved byte for byte. An identical completed send replays Cave's + * recorded result; the SDK never replays one on its own. + */ + async sendConversationMessage( + conversationId: string, + request: CaveSendConversationMessageRequest, + options: OperationOptions = {}, + ): Promise { + const validatedConversationId = validateCanonicalId( + conversationId, + 'conversationId', + ); + const validated = parseSendConversationMessageRequest(request); + + return await this.#conversationMutation( + 'sendConversationMessage', + validated.operationId, + options, + async (context) => { + const call = this.#transport.sendConversationMessage?.bind(this.#transport); + const response = this.#managedSnapshot( + await this.#conversationTransportCall( + 'sendConversationMessage', + call === undefined + ? undefined + : () => call(validatedConversationId, validated, context), + ), + 'sendConversationMessage', + ); + return this.#parseConversationResponse( + 'sendConversationMessage', + () => parseSendConversationMessageResult(response, validated.operationId), + ); + }, + ); + } + + /** + * Typed convenience over `messages.send` for retrying an explicitly failed + * or cancelled assistant turn. It uses a fresh operation UUID and the + * explicit `retryOfTurnId`; it introduces no second producer route. + */ + async retryConversationTurn( + conversationId: string, + request: CaveRetryConversationTurnRequest, + options: OperationOptions = {}, + ): Promise { + const validated = parseRetryConversationTurnRequest(request); + return await this.sendConversationMessage( + conversationId, + { operationId: validated.operationId, retryOfTurnId: validated.retryOfTurnId }, + options, + ); + } + + /** + * The non-content operation record: fixed codes, turn references, event + * bounds, and timestamps. Prompt, attachment, bearer, and raw-cause + * content never appear here. + */ + async getConversationOperation( + operationId: string, + options: OperationOptions = {}, + ): Promise { + const validatedId = validateConversationOperationId(operationId); + + return this.#execute('getConversationOperation', options, async (context) => { + this.#ensureActive(context, 'getConversationOperation'); + const call = this.#transport.getConversationOperation?.bind(this.#transport); + const response = this.#managedSnapshot( + await this.#conversationTransportCall( + 'getConversationOperation', + call === undefined ? undefined : () => call(validatedId, context), + ), + 'getConversationOperation', + ); + try { + return this.#parseConversationResponse('getConversationOperation', () => + parseConversationOperationResponse(response, validatedId), + ); + } catch (error) { + if (error instanceof CaveConversationSchemaError) { + throw invalidCanonicalResponse('getConversationOperation', error.field); + } + throw error; + } + }, true, this.#usesManagedCredentialTransport()); + } + + /** + * The typed, resumable event stream for one conversation operation. + * + * `options.timeoutMs` is one total stream budget; each long poll receives + * only the remaining budget. A caller abort closes the current event read + * and this generator: it never calls Stop and never resubmits a send. + * Initial attachment and every resumed page pass through the same event + * translator, which suppresses duplicates at or below the accepted cursor + * and refuses protocol violations. `reconcile_required` is an instruction + * to reload canonical history, not a retry. + */ + streamConversationOperation( + operationId: string, + options: CaveConversationStreamOptions = {}, + ): AsyncGenerator { + return this.#streamConversationEvents(operationId, options); + } + + async *#streamConversationEvents( + operationId: string, + options: CaveConversationStreamOptions, + ): AsyncGenerator { + const validatedId = validateConversationOperationId(operationId); + const resumeCursor = + options.cursor === undefined + ? undefined + : validateConversationEventCursor(options.cursor, 'cursor'); + const translator = createConversationEventTranslator(validatedId, { + ...(resumeCursor === undefined ? {} : { resumeAfterOpaqueCursor: true }), + }); + const callerAborted = (): boolean => options.signal?.aborted === true; + const totalBudgetMs = options.timeoutMs ?? this.#operation?.timeoutMs; + const deadline = + totalBudgetMs === undefined ? undefined : performance.now() + totalBudgetMs; + let cursor: CaveConversationEventCursor | undefined = resumeCursor; + + const readPage = async (): Promise => { + let remainingMs: number | undefined; + if (deadline !== undefined) { + remainingMs = Math.floor(deadline - performance.now()); + if (remainingMs < 1) { + const cause = { code: 'timeout', retryable: true }; + throw attachConversationOperationId( + new CaveClientError( + normalizeCaveError(cause, 'streamConversationOperation'), + undefined, + { cause }, + ), + validatedId, + ); + } + } + + try { + const page = await this.#execute( + 'streamConversationOperation', + { + ...(options.signal === undefined ? {} : { signal: options.signal }), + ...(remainingMs === undefined ? {} : { timeoutMs: remainingMs }), + ...(options.observer === undefined ? {} : { observer: options.observer }), + }, + async (context) => { + this.#ensureActive(context, 'streamConversationOperation'); + const call = this.#transport.readConversationOperationEvents?.bind( + this.#transport, + ); + if (call === undefined) { + throw unsupported('streamConversationOperation'); + } + return this.#managedSnapshot( + await call( + validatedId, + { ...(cursor === undefined ? {} : { cursor }) }, + context, + ), + 'streamConversationOperation', + ); + }, + true, + this.#usesManagedCredentialTransport(), + ); + try { + return this.#parseConversationResponse('streamConversationOperation', () => + translator.translate(page), + ); + } catch (error) { + if (error instanceof CaveConversationSchemaError) { + throw invalidCanonicalResponse('streamConversationOperation', error.field); + } + // Route errors arrive as typed response errors; surface them as + // normalized client errors with their bounded details intact. + if (error instanceof CaveConversationResponseError) { + const cause = error; + throw attachConversationOperationId( + new CaveClientError( + normalizeCaveError(error, 'streamConversationOperation'), + undefined, + { cause }, + ), + validatedId, + ); + } + throw error; + } + } catch (error) { + throw attachConversationOperationId(error, validatedId); + } + }; + + while (true) { + if (callerAborted()) { + // A caller abort closes the current event read and this generator. + // It never calls Stop and never resubmits a send. + return; + } + + let page: CaveConversationTranslatedPage; + try { + page = await readPage(); + } catch (error) { + if ( + isOperationAbortedError(error) || + (isCaveClientError(error) && error.code === 'aborted') + ) { + // A caller abort closes the current event read and this generator. + // It never calls Stop and never resubmits a send. + return; + } + throw error; + } + + for (const event of page.events) { + // A caller abort closes the current event read and this generator + // immediately: buffered events are not delivered and the resume + // cursor stays at the last delivered event. + if (callerAborted()) { + return; + } + // The resume cursor advances only after the event is delivered. + yield event; + translator.commit(event.eventId); + cursor = event.cursor; + } + + if (page.complete) { + return; + } + if (page.nextCursor !== undefined) { + cursor = page.nextCursor; + } + // An empty page with `complete: false` keeps polling the same cursor. + } + } + + /** + * Explicit Stop for one conversation operation. Repeated Stop calls are + * safe against the target operation; this client sends each Stop exactly + * once and never retries it after ambiguous transport completion. + */ + async stopConversationOperation( + operationId: string, + options: OperationOptions = {}, + ): Promise { + const validatedId = validateConversationOperationId(operationId); + + return await this.#conversationMutation( + 'stopConversationOperation', + validatedId, + options, + async (context) => { + const call = this.#transport.stopConversationOperation?.bind(this.#transport); + const response = this.#managedSnapshot( + await this.#conversationTransportCall( + 'stopConversationOperation', + call === undefined ? undefined : () => call(validatedId, context), + ), + 'stopConversationOperation', + ); + return this.#parseConversationResponse( + 'stopConversationOperation', + () => parseConversationOperationResponse(response, validatedId), + ); + }, + ); + } } export function createCaveClient(options: CaveClientOptions): CaveClient { diff --git a/packages/cave/src/contract-constraints.ts b/packages/cave/src/contract-constraints.ts index 3149953..1f78671 100644 --- a/packages/cave/src/contract-constraints.ts +++ b/packages/cave/src/contract-constraints.ts @@ -17,6 +17,7 @@ export const CAVE_CONTRACT_ERROR_CODES = [ ] as const; export const CAVE_CONTRACT_LIMITS = Object.freeze({ + cursorCharacters: 512, declarationIdCharacters: 64, errorDetailEntries: 16, errorDetailValueCharacters: 256, diff --git a/packages/cave/src/conversation-control.ts b/packages/cave/src/conversation-control.ts new file mode 100644 index 0000000..b49b0d9 --- /dev/null +++ b/packages/cave/src/conversation-control.ts @@ -0,0 +1,1235 @@ +import { + assessCompatibility, + OperationConfigurationError, + type OperationObserver, +} from '@opencoven/sdk-core/browser'; + +import { parseConversation } from './canonical-reads.js'; +import { + CAVE_CONTRACT_API_VERSION, + CAVE_CONTRACT_LIMITS, + isCaveContractErrorCode, +} from './contract-constraints.js'; +import type { CaveConversation } from './schemas.js'; +import { CAVE_CLIENT_VERSION } from './version.js'; + +/** + * Conversational control: the first bounded mutation authority. + * + * Cave remains the sole executor and canonical state owner. The SDK exposes + * constrained typed operations only — never arbitrary HTTP paths, private + * Cave routes, or raw transport escape hatches — and owns the public DTOs, + * validators, and the single event translator shared by initial and resumed + * streams. + * + * The five Client v1 operations this surface is defined against + * (`conversations.create`, `messages.send`, `operations.read`, + * `operations.events`, `operations.stop`) are not yet declared by the + * authoritative Cave contract fixture this SDK vendors. This module therefore + * defines the typed requests, results, operation records, event vocabulary, + * cursor handling, and translation rules only; it introduces no HTTP paths. + * Transport bindings for the five operations stay optional and are expected + * to arrive with the upstream Cave producer contract and a re-imported + * fixture. + * + * This module is import-pure: no discovery, credential, filesystem, network, + * or daemon I/O happens at import time. + */ + +export type CaveConversationOperationId = string; + +export type CaveConversationEventCursor = string; + +export interface CaveCreateConversationRequest { + operationId: CaveConversationOperationId; + familiarId: string; + projectId?: string; +} + +export type CaveSendConversationMessageRequest = + | { + operationId: CaveConversationOperationId; + text: string; + retryOfTurnId?: never; + } + | { + operationId: CaveConversationOperationId; + retryOfTurnId: string; + text?: never; + }; + +export interface CaveRetryConversationTurnRequest { + operationId: CaveConversationOperationId; + retryOfTurnId: string; +} + +export type CaveConversationOperationState = + | 'pending' + | 'accepted' + | 'running' + | 'stopping' + | 'completed' + | 'failed' + | 'cancelled'; + +export type CaveConversationOperationKind = + | 'conversations.create' + | 'messages.send'; + +export type CaveConversationOriginatingScope = + | 'chat:write' + | 'conversations:write'; + +export interface CaveConversationOperation { + id: CaveConversationOperationId; + kind: CaveConversationOperationKind; + state: CaveConversationOperationState; + originatingScope: CaveConversationOriginatingScope; + conversationId: string; + inputTurnId?: string; + outputTurnId?: string; + retryOfTurnId?: string; + failureCode?: string; + latestEventId: number; + replayFloorEventId: number; + createdAt: string; + updatedAt: string; + idempotencyResultExpiresAt?: string; +} + +export interface CaveCreateConversationResult { + operationId: CaveConversationOperationId; + replayed: boolean; + conversation: CaveConversation; +} + +export interface CaveSendConversationMessageResult { + operation: CaveConversationOperation; + replayed: boolean; +} + +export interface CaveConversationEventBase { + operationId: CaveConversationOperationId; + eventId: number; + cursor: CaveConversationEventCursor; + occurredAt: string; +} + +export type CaveConversationEventType = + | 'operation.accepted' + | 'assistant.delta' + | 'operation.stopping' + | 'operation.completed' + | 'operation.failed' + | 'operation.cancelled'; + +export type CaveConversationEvent = + | (CaveConversationEventBase & { + type: 'operation.accepted'; + conversationId: string; + inputTurnId: string; + retryOfTurnId?: string; + }) + | (CaveConversationEventBase & { + type: 'assistant.delta'; + text: string; + }) + | (CaveConversationEventBase & { + type: 'operation.stopping'; + }) + | (CaveConversationEventBase & { + type: 'operation.completed'; + outputTurnId: string; + }) + | (CaveConversationEventBase & { + type: 'operation.failed'; + outputTurnId: string; + code: string; + }) + | (CaveConversationEventBase & { + type: 'operation.cancelled'; + outputTurnId: string; + }); + +export interface CaveConversationEventPage { + operation: CaveConversationOperation; + events: readonly CaveConversationEvent[]; + complete: boolean; + cursor?: { + current?: CaveConversationEventCursor; + next?: CaveConversationEventCursor; + hasMore: boolean; + }; +} + +export interface CaveConversationEventPageRequest { + cursor?: CaveConversationEventCursor; + waitMs?: number; +} + +export interface CaveConversationStreamOptions { + cursor?: CaveConversationEventCursor; + signal?: AbortSignal; + timeoutMs?: number; + observer?: OperationObserver; +} + +export const CAVE_CONVERSATION_OPERATION_STATES = [ + 'pending', + 'accepted', + 'running', + 'stopping', + 'completed', + 'failed', + 'cancelled', +] as const; + +export const CAVE_CONVERSATION_TERMINAL_STATES = [ + 'completed', + 'failed', + 'cancelled', +] as const; + +export const CAVE_CONVERSATION_EVENT_TYPES = [ + 'operation.accepted', + 'assistant.delta', + 'operation.stopping', + 'operation.completed', + 'operation.failed', + 'operation.cancelled', +] as const; + +/** The scope stored with an operation when it was claimed; reads of the operation and its events are authorized by it. */ +export const CAVE_CONVERSATION_ORIGINATING_SCOPES = [ + 'chat:write', + 'conversations:write', +] as const; + +/** + * The defined `reconcile_required` reasons. A `reconcile_required` error is + * an instruction to reload canonical state, not a transient transport retry. + */ +export const CAVE_CONVERSATION_RECONCILE_REASONS = [ + 'replay_gap', + 'operation_expired', + 'canonical_branch_changed', + 'idempotency_result_expired', + 'canonical_state_moved', +] as const; + +export type CaveConversationReconcileReason = + (typeof CAVE_CONVERSATION_RECONCILE_REASONS)[number]; + +export class CaveConversationSchemaError extends TypeError { + readonly field: string; + + constructor(field: string) { + super(`${field} was malformed.`); + this.name = 'CaveConversationSchemaError'; + this.field = field; + } +} + +/** + * A contract error envelope (`error` inside the shared Client v1 envelope). + * `details` carries bounded string-valued route metadata only — for example + * the `reconcile_required` reason. + */ +export class CaveConversationResponseError extends Error { + readonly code: string; + readonly retryable: boolean; + readonly details: Record | undefined; + readonly requestId: string | undefined; + + constructor( + code: string, + message: string, + options: { + details?: Record; + requestId?: string; + retryable?: boolean; + } = {}, + ) { + super(message); + this.name = 'CaveConversationResponseError'; + this.code = code; + this.retryable = options.retryable ?? false; + this.details = options.details; + this.requestId = options.requestId; + } +} + +type JsonObject = Record; + +const CONVERSATION_STATE_SET = new Set(CAVE_CONVERSATION_OPERATION_STATES); +const CONVERSATION_TERMINAL_STATE_SET = new Set( + CAVE_CONVERSATION_TERMINAL_STATES, +); +const CONVERSATION_EVENT_TYPE_SET = new Set(CAVE_CONVERSATION_EVENT_TYPES); + +/** + * The existing Client v1 UUID contract: exactly 36 characters and the + * current RFC-compatible UUID pattern. Cave normalizes accepted UUIDs to + * lowercase before key lookup, so case variants cannot claim two operations. + */ +const CONVERSATION_UUID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/iu; + +const DECLARATION_ID_PATTERN = /^[a-z][a-z0-9]*(?:[.-][a-z0-9]+)*$/u; + +function conversationObject(value: unknown, field: string): JsonObject { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new CaveConversationSchemaError(field); + } + return value as JsonObject; +} + +function conversationString(value: unknown, field: string): string { + if (typeof value !== 'string') { + throw new CaveConversationSchemaError(field); + } + return value; +} + +function conversationBoundedString( + value: unknown, + field: string, + maximumLength: number, + options: { requireNonEmpty?: boolean } = {}, +): string { + const parsed = conversationString(value, field); + if ( + parsed.length > maximumLength || + (options.requireNonEmpty === true && parsed.length === 0) + ) { + throw new CaveConversationSchemaError(field); + } + return parsed; +} + +function conversationBoolean(value: unknown, field: string): boolean { + if (typeof value !== 'boolean') { + throw new CaveConversationSchemaError(field); + } + return value; +} + +function conversationCount(value: unknown, field: string): number { + if (!Number.isSafeInteger(value) || (value as number) < 0) { + throw new CaveConversationSchemaError(field); + } + return value as number; +} + +function conversationOptionalNonEmptyString( + value: unknown, + field: string, +): string | undefined { + return value === undefined + ? undefined + : conversationBoundedString( + value, + field, + CAVE_CONTRACT_LIMITS.errorMessageCharacters, + { requireNonEmpty: true }, + ); +} + +function conversationExactKeys( + value: JsonObject, + allowed: ReadonlySet, + field: string, +): void { + for (const key of Object.keys(value)) { + if (!allowed.has(key)) { + throw new CaveConversationSchemaError(`${field}.${key}`); + } + } +} + +function conversationDeclarationIds(value: unknown, field: string): string[] { + if (!Array.isArray(value) || value.length === 0) { + throw new CaveConversationSchemaError(field); + } + const declarations: string[] = []; + for (const [index, entry] of value.entries()) { + const declaration = conversationBoundedString( + entry, + `${field}[${index}]`, + CAVE_CONTRACT_LIMITS.declarationIdCharacters, + { requireNonEmpty: true }, + ); + if (!DECLARATION_ID_PATTERN.test(declaration) || declarations.includes(declaration)) { + throw new CaveConversationSchemaError(`${field}[${index}]`); + } + declarations.push(declaration); + } + return declarations; +} + +function requireOwnShape( + value: unknown, + label: string, +): { record: JsonObject; keys: Set } { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new OperationConfigurationError(`${label} must be an object`); + } + const record = value as JsonObject; + return { record, keys: new Set(Object.keys(record)) }; +} + +function rejectUnknownRequestKeys( + present: ReadonlySet, + allowed: ReadonlySet, + label: string, +): void { + for (const key of present) { + if (!allowed.has(key)) { + throw new OperationConfigurationError(`${label} has an unknown field`); + } + } +} + +function requireRequestKeys( + present: ReadonlySet, + required: readonly string[], + label: string, +): void { + for (const key of required) { + if (!present.has(key)) { + throw new OperationConfigurationError(`${label} requires ${key}`); + } + } +} + +/** + * The caller-supplied, caller-visible operation UUID. Exactly 36 characters, + * RFC-compatible, normalized to lowercase so case variants cannot claim two + * operations. The untrusted value is never echoed in the error. + */ +export function validateConversationOperationId( + value: unknown, +): CaveConversationOperationId { + if ( + typeof value !== 'string' || + value.length !== 36 || + !CONVERSATION_UUID_PATTERN.test(value) + ) { + throw new OperationConfigurationError( + 'operationId must be a 36-character UUID', + ); + } + return value.toLowerCase(); +} + +/** + * Event cursors are opaque route strings bounded by the authoritative + * `cursorCharacters` limit. The SDK never decodes them. + */ +export function validateConversationEventCursor( + value: unknown, + label: string, +): CaveConversationEventCursor { + if (typeof value !== 'string' || value.length === 0) { + throw new OperationConfigurationError(`${label} must be a non-empty string`); + } + if (value.length > CAVE_CONTRACT_LIMITS.cursorCharacters) { + throw new OperationConfigurationError( + `${label} must be at most ${CAVE_CONTRACT_LIMITS.cursorCharacters} characters`, + ); + } + return value; +} + +function validateConversationTargetId(value: unknown, label: string): string { + if (typeof value !== 'string' || value.trim().length === 0) { + throw new OperationConfigurationError(`${label} must be a non-empty string`); + } + if (value === '.' || value === '..') { + throw new OperationConfigurationError( + `${label} must not be a dot path segment`, + ); + } + return value; +} + +const CREATE_REQUEST_KEYS = new Set(['operationId', 'familiarId', 'projectId']); +const SEND_REQUEST_KEYS = new Set(['operationId', 'text', 'retryOfTurnId']); +const RETRY_REQUEST_KEYS = new Set(['operationId', 'retryOfTurnId']); + +/** + * Create accepts only the operation UUID, one canonical familiar ID, and an + * optional canonical project ID. Cave resolves project roots, harnesses, + * runtimes, titles, and origin internally: the caller cannot send a + * filesystem path, harness command, runtime URL, model-provider payload, + * origin marker, or prebuilt transcript. + */ +export function parseCreateConversationRequest( + value: unknown, +): CaveCreateConversationRequest { + const { record, keys } = requireOwnShape(value, 'createConversation request'); + rejectUnknownRequestKeys(keys, CREATE_REQUEST_KEYS, 'createConversation request'); + requireRequestKeys(keys, ['operationId', 'familiarId'], 'createConversation request'); + const operationId = validateConversationOperationId(record.operationId); + const familiarId = validateConversationTargetId(record.familiarId, 'familiarId'); + const projectId = + record.projectId === undefined + ? undefined + : validateConversationTargetId(record.projectId, 'projectId'); + + return { + operationId, + familiarId, + ...(projectId === undefined ? {} : { projectId }), + }; +} + +/** + * A send request carries exactly one of `text` (a new send) or + * `retryOfTurnId` (an explicit retry of a failed or cancelled assistant + * turn). Text is preserved byte for byte: Cave does not Unicode-normalize or + * trim persisted text, and the canonical request hash distinguishes + * whitespace and normalization variants. + */ +export function parseSendConversationMessageRequest( + value: unknown, +): CaveSendConversationMessageRequest { + const { record, keys } = requireOwnShape( + value, + 'sendConversationMessage request', + ); + rejectUnknownRequestKeys(keys, SEND_REQUEST_KEYS, 'sendConversationMessage request'); + requireRequestKeys(keys, ['operationId'], 'sendConversationMessage request'); + const operationId = validateConversationOperationId(record.operationId); + const hasText = record.text !== undefined; + const hasRetry = record.retryOfTurnId !== undefined; + + if (hasText === hasRetry) { + throw new OperationConfigurationError( + 'sendConversationMessage requires exactly one of text or retryOfTurnId', + ); + } + + if (hasRetry) { + const retryOfTurnId = validateConversationTargetId( + record.retryOfTurnId, + 'retryOfTurnId', + ); + return { operationId, retryOfTurnId }; + } + + const text = record.text; + if (typeof text !== 'string' || text.trim().length === 0) { + throw new OperationConfigurationError( + 'sendConversationMessage text must be a non-empty string', + ); + } + return { operationId, text }; +} + +/** + * Retry requires a fresh operation UUID and the canonical `retryOfTurnId`. + * It carries no replacement text: the executor reads the exact canonical + * parent user content from the transcript. + */ +export function parseRetryConversationTurnRequest( + value: unknown, +): CaveRetryConversationTurnRequest { + const { record, keys } = requireOwnShape( + value, + 'retryConversationTurn request', + ); + rejectUnknownRequestKeys(keys, RETRY_REQUEST_KEYS, 'retryConversationTurn request'); + requireRequestKeys(keys, ['operationId', 'retryOfTurnId'], 'retryConversationTurn request'); + const operationId = validateConversationOperationId(record.operationId); + const retryOfTurnId = validateConversationTargetId( + record.retryOfTurnId, + 'retryOfTurnId', + ); + return { operationId, retryOfTurnId }; +} + +const OPERATION_DTO_KEYS = new Set([ + 'id', + 'kind', + 'state', + 'originatingScope', + 'conversationId', + 'inputTurnId', + 'outputTurnId', + 'retryOfTurnId', + 'failureCode', + 'latestEventId', + 'replayFloorEventId', + 'createdAt', + 'updatedAt', + 'idempotencyResultExpiresAt', +]); + +/** + * The non-content operation record: fixed codes, turn references, event + * bounds, and timestamps only. Prompt, attachment, bearer, pairing secret, + * HPKE material, stack, cause, command output, or environment content never + * appears here. + */ +export function parseConversationOperation( + value: unknown, + field: string, +): CaveConversationOperation { + const operation = conversationObject(value, field); + conversationExactKeys(operation, OPERATION_DTO_KEYS, field); + + const id = conversationString(operation.id, `${field}.id`); + if ( + id.length !== 36 || + !CONVERSATION_UUID_PATTERN.test(id) || + id !== id.toLowerCase() + ) { + throw new CaveConversationSchemaError(`${field}.id`); + } + + const kind = conversationString(operation.kind, `${field}.kind`); + if (kind !== 'conversations.create' && kind !== 'messages.send') { + throw new CaveConversationSchemaError(`${field}.kind`); + } + const state = conversationString(operation.state, `${field}.state`); + if (!CONVERSATION_STATE_SET.has(state)) { + throw new CaveConversationSchemaError(`${field}.state`); + } + const originatingScope = conversationString( + operation.originatingScope, + `${field}.originatingScope`, + ); + if (originatingScope !== 'chat:write' && originatingScope !== 'conversations:write') { + throw new CaveConversationSchemaError(`${field}.originatingScope`); + } + const conversationId = conversationBoundedString( + operation.conversationId, + `${field}.conversationId`, + CAVE_CONTRACT_LIMITS.cursorCharacters, + { requireNonEmpty: true }, + ); + const inputTurnId = conversationOptionalNonEmptyString( + operation.inputTurnId, + `${field}.inputTurnId`, + ); + const outputTurnId = conversationOptionalNonEmptyString( + operation.outputTurnId, + `${field}.outputTurnId`, + ); + const retryOfTurnId = conversationOptionalNonEmptyString( + operation.retryOfTurnId, + `${field}.retryOfTurnId`, + ); + const failureCode = conversationOptionalNonEmptyString( + operation.failureCode, + `${field}.failureCode`, + ); + const latestEventId = conversationCount( + operation.latestEventId, + `${field}.latestEventId`, + ); + const replayFloorEventId = conversationCount( + operation.replayFloorEventId, + `${field}.replayFloorEventId`, + ); + const createdAt = conversationBoundedString( + operation.createdAt, + `${field}.createdAt`, + CAVE_CONTRACT_LIMITS.errorMessageCharacters, + { requireNonEmpty: true }, + ); + const updatedAt = conversationBoundedString( + operation.updatedAt, + `${field}.updatedAt`, + CAVE_CONTRACT_LIMITS.errorMessageCharacters, + { requireNonEmpty: true }, + ); + const idempotencyResultExpiresAt = conversationOptionalNonEmptyString( + operation.idempotencyResultExpiresAt, + `${field}.idempotencyResultExpiresAt`, + ); + + if (replayFloorEventId < 1 || replayFloorEventId > latestEventId + 1) { + throw new CaveConversationSchemaError(`${field}.replayFloorEventId`); + } + if ( + CONVERSATION_TERMINAL_STATE_SET.has(state) && + (latestEventId < 1 || outputTurnId === undefined) + ) { + throw new CaveConversationSchemaError(`${field}.latestEventId`); + } + + return { + id, + kind: kind, + state: state as CaveConversationOperationState, + originatingScope: originatingScope, + conversationId, + ...(inputTurnId === undefined ? {} : { inputTurnId }), + ...(outputTurnId === undefined ? {} : { outputTurnId }), + ...(retryOfTurnId === undefined ? {} : { retryOfTurnId }), + ...(failureCode === undefined ? {} : { failureCode }), + latestEventId, + replayFloorEventId, + createdAt, + updatedAt, + ...(idempotencyResultExpiresAt === undefined + ? {} + : { idempotencyResultExpiresAt }), + }; +} + +const ERROR_ENVELOPE_KEYS = new Set(['code', 'message', 'retryable', 'details']); + +function parseConversationErrorDetails( + value: unknown, +): Record | undefined { + if (value === undefined) { + return undefined; + } + const details = conversationObject(value, 'error.details'); + const entries = Object.entries(details); + if (entries.length > CAVE_CONTRACT_LIMITS.errorDetailEntries) { + throw new CaveConversationSchemaError('error.details'); + } + return Object.fromEntries( + entries.map(([key, entry]) => [ + key, + conversationBoundedString( + entry, + `error.details.${key}`, + CAVE_CONTRACT_LIMITS.errorDetailValueCharacters, + { requireNonEmpty: true }, + ), + ]), + ); +} + +/** + * The shared Client v1 envelope: apiVersion, compatibility, declaration + * metadata, requestId bounds, and exactly one of `data` or `error`. A well- + * formed error envelope becomes the typed route error so fixed codes and + * bounded details (for example the `reconcile_required` reason) survive. + * + * The mutation contract's operation and capability declarations are not yet + * declared by the authoritative fixture, so the envelope's declarations are + * validated as declarations but not pinned to specific conversation + * operation identifiers here. That pinning is an upstream-contract gap. + */ +function parseConversationEnvelopeMetadata( + value: unknown, +): { envelope: JsonObject; requestId: string | undefined } { + const envelope = conversationObject(value, 'response'); + const apiVersion = conversationString(envelope.apiVersion, 'response.apiVersion'); + if (apiVersion !== CAVE_CONTRACT_API_VERSION) { + throw new CaveConversationSchemaError('response.apiVersion'); + } + const minimumClientVersion = conversationString( + envelope.minimumClientVersion, + 'response.minimumClientVersion', + ); + conversationDeclarationIds(envelope.capabilities, 'response.capabilities'); + conversationDeclarationIds(envelope.operations, 'response.operations'); + + let compatible: boolean; + try { + compatible = assessCompatibility(minimumClientVersion, CAVE_CLIENT_VERSION).compatible; + } catch { + throw new CaveConversationSchemaError('response.minimumClientVersion'); + } + if (!compatible) { + throw new CaveConversationResponseError( + 'incompatible_version', + 'Cave minimumClientVersion was not compatible.', + ); + } + + const requestId = + envelope.requestId === undefined + ? undefined + : conversationBoundedString( + envelope.requestId, + 'response.requestId', + CAVE_CONTRACT_LIMITS.requestIdCharacters, + { requireNonEmpty: true }, + ); + + const hasData = envelope.data !== undefined; + const hasError = envelope.error !== undefined; + if (hasData === hasError) { + throw new CaveConversationSchemaError('response'); + } + + if (hasError) { + const error = conversationObject(envelope.error, 'error'); + conversationExactKeys(error, ERROR_ENVELOPE_KEYS, 'error'); + const code = conversationString(error.code, 'error.code'); + if (!isCaveContractErrorCode(code)) { + throw new CaveConversationSchemaError('error.code'); + } + const message = conversationBoundedString( + error.message, + 'error.message', + CAVE_CONTRACT_LIMITS.errorMessageCharacters, + { requireNonEmpty: true }, + ); + const retryable = conversationBoolean(error.retryable, 'error.retryable'); + const details = parseConversationErrorDetails(error.details); + throw new CaveConversationResponseError(code, message, { + retryable, + ...(details === undefined ? {} : { details }), + ...(requestId === undefined ? {} : { requestId }), + }); + } + + return { envelope, requestId }; +} + +const CREATE_RESULT_KEYS = new Set(['operationId', 'replayed', 'conversation']); +const SEND_RESULT_KEYS = new Set(['operation', 'replayed']); +const OPERATION_RESULT_KEYS = new Set(['operation']); + +/** The recorded create result: operation UUID, replay flag, canonical conversation. */ +export function parseCreateConversationResult( + value: unknown, + expectedOperationId: CaveConversationOperationId, +): CaveCreateConversationResult { + const { envelope } = parseConversationEnvelopeMetadata(value); + const data = conversationObject(envelope.data, 'data'); + conversationExactKeys(data, CREATE_RESULT_KEYS, 'data'); + + const operationId = validateConversationOperationId(data.operationId); + if (operationId !== expectedOperationId) { + throw new CaveConversationSchemaError('data.operationId'); + } + const replayed = conversationBoolean(data.replayed, 'data.replayed'); + const conversation = parseConversation(data.conversation, 'data.conversation'); + + return { operationId, replayed, conversation }; +} + +/** The send/retry acceptance result: the claimed operation and replay flag. */ +export function parseSendConversationMessageResult( + value: unknown, + expectedOperationId: CaveConversationOperationId, +): CaveSendConversationMessageResult { + const { envelope } = parseConversationEnvelopeMetadata(value); + const data = conversationObject(envelope.data, 'data'); + conversationExactKeys(data, SEND_RESULT_KEYS, 'data'); + const operation = parseConversationOperation(data.operation, 'data.operation'); + if (operation.id !== expectedOperationId) { + throw new CaveConversationSchemaError('data.operation.id'); + } + const replayed = conversationBoolean(data.replayed, 'data.replayed'); + return { operation, replayed }; +} + +/** One non-content operation record, as read and stop routes return it. */ +export function parseConversationOperationResponse( + value: unknown, + expectedOperationId: CaveConversationOperationId, +): CaveConversationOperation { + const { envelope } = parseConversationEnvelopeMetadata(value); + const data = conversationObject(envelope.data, 'data'); + conversationExactKeys(data, OPERATION_RESULT_KEYS, 'data'); + const operation = parseConversationOperation(data.operation, 'data.operation'); + if (operation.id !== expectedOperationId) { + throw new CaveConversationSchemaError('data.operation.id'); + } + return operation; +} + +const EVENT_PAYLOAD_KEYS: Record> = { + 'operation.accepted': new Set(['conversationId', 'inputTurnId', 'retryOfTurnId']), + 'assistant.delta': new Set(['text']), + 'operation.stopping': new Set([]), + 'operation.completed': new Set(['outputTurnId']), + 'operation.failed': new Set(['outputTurnId', 'code']), + 'operation.cancelled': new Set(['outputTurnId']), +}; + +function parseConversationEvent( + value: unknown, + expectedOperationId: string, + field: string, +): CaveConversationEvent { + const event = conversationObject(value, field); + const type = conversationString(event.type, `${field}.type`); + if (!CONVERSATION_EVENT_TYPE_SET.has(type)) { + throw new CaveConversationSchemaError(`${field}.type`); + } + const allowed = new Set([ + 'type', + 'operationId', + 'eventId', + 'cursor', + 'occurredAt', + ]); + const payloadKeys = EVENT_PAYLOAD_KEYS[type]; + if (payloadKeys === undefined) { + throw new CaveConversationSchemaError(`${field}.type`); + } + for (const key of payloadKeys) { + allowed.add(key); + } + conversationExactKeys(event, allowed, field); + + if ( + conversationString(event.operationId, `${field}.operationId`) !== + expectedOperationId + ) { + throw new CaveConversationSchemaError(`${field}.operationId`); + } + const eventId = conversationCount(event.eventId, `${field}.eventId`); + if (eventId < 1) { + throw new CaveConversationSchemaError(`${field}.eventId`); + } + const cursor = conversationBoundedString( + event.cursor, + `${field}.cursor`, + CAVE_CONTRACT_LIMITS.cursorCharacters, + { requireNonEmpty: true }, + ); + const occurredAt = conversationBoundedString( + event.occurredAt, + `${field}.occurredAt`, + CAVE_CONTRACT_LIMITS.requestIdCharacters, + { requireNonEmpty: true }, + ); + const base = { + type: type as CaveConversationEventType, + operationId: expectedOperationId, + eventId, + cursor, + occurredAt, + }; + + if (type === 'assistant.delta') { + return { ...base, type, text: conversationString(event.text, `${field}.text`) }; + } + if (type === 'operation.stopping') { + return { ...base, type }; + } + + if (type === 'operation.accepted') { + const conversationId = conversationBoundedString( + event.conversationId, + `${field}.conversationId`, + CAVE_CONTRACT_LIMITS.cursorCharacters, + { requireNonEmpty: true }, + ); + const inputTurnId = conversationBoundedString( + event.inputTurnId, + `${field}.inputTurnId`, + CAVE_CONTRACT_LIMITS.cursorCharacters, + { requireNonEmpty: true }, + ); + const retryOfTurnId = conversationOptionalNonEmptyString( + event.retryOfTurnId, + `${field}.retryOfTurnId`, + ); + return { + ...base, + type, + conversationId, + inputTurnId, + ...(retryOfTurnId === undefined ? {} : { retryOfTurnId }), + }; + } + const outputTurnId = conversationBoundedString( + event.outputTurnId, + `${field}.outputTurnId`, + CAVE_CONTRACT_LIMITS.cursorCharacters, + { requireNonEmpty: true }, + ); + if (type === 'operation.failed') { + const code = conversationBoundedString( + event.code, + `${field}.code`, + CAVE_CONTRACT_LIMITS.declarationIdCharacters, + { requireNonEmpty: true }, + ); + return { ...base, type, outputTurnId, code }; + } + if (type === 'operation.completed' || type === 'operation.cancelled') { + return { ...base, type, outputTurnId }; + } + // The remaining event kinds were all handled above; an event whose type + // passed the allowlist cannot reach here. + throw new CaveConversationSchemaError(`${field}.type`); +} + +function isTerminalEventType(event: CaveConversationEvent): boolean { + return ( + event.type === 'operation.completed' || + event.type === 'operation.failed' || + event.type === 'operation.cancelled' + ); +} + +export interface CaveConversationTranslatedPage { + operation: CaveConversationOperation; + events: readonly CaveConversationEvent[]; + complete: boolean; + requestId: string | undefined; + nextCursor?: CaveConversationEventCursor; +} + +const PAGE_DATA_KEYS = new Set(['operation', 'events', 'complete', 'cursor']); +const PAGE_CURSOR_KEYS = new Set(['current', 'next', 'hasMore']); + +export interface CaveConversationEventTranslator { + readonly operationId: CaveConversationOperationId; + readonly deliveredThroughEventId: number; + /** + * Validate one raw event-page response and return the accepted events in + * wire order. Throws a protocol error on any violation. + */ + translate(page: unknown): CaveConversationTranslatedPage; + /** Advance the accepted cursor after the event has been delivered. */ + commit(eventId: number): void; +} + +export interface CaveConversationEventTranslatorOptions { + /** + * True when this stream's first page resumes behind an opaque cursor from + * an earlier generator run: the first accepted event then cannot be + * gap-checked against a known event ID. A fresh stream must begin at + * event 1. + */ + resumeAfterOpaqueCursor?: boolean; +} + +/** + * The one parser/translator for conversation event pages. Initial attachment + * and every resumed long poll pass through it. + * + * The translator validates the shared Client v1 envelope before event data, + * validates the operation ID on every event, requires contiguous increasing + * event IDs, suppresses an exact duplicate event at or below the caller's + * accepted cursor, and refuses forward gaps, reordered events, changed + * operation IDs, malformed terminal sequences, and events after terminal as + * protocol violations. The resume cursor advances only when the caller + * commits, after the corresponding event has been delivered. + */ +export function createConversationEventTranslator( + operationId: CaveConversationOperationId, + options: CaveConversationEventTranslatorOptions = {}, +): CaveConversationEventTranslator { + const operationIdCanonical = validateConversationOperationId(operationId); + let deliveredThrough = 0; + let firstPage = true; + let sawStopping = false; + let sawTerminal = false; + + return { + get operationId(): CaveConversationOperationId { + return operationIdCanonical; + }, + get deliveredThroughEventId(): number { + return deliveredThrough; + }, + commit(eventId: number): void { + if (!Number.isSafeInteger(eventId) || eventId < 1 || eventId < deliveredThrough) { + throw new OperationConfigurationError( + 'commit must advance the accepted cursor monotonically', + ); + } + deliveredThrough = eventId; + }, + translate(page: unknown): CaveConversationTranslatedPage { + if (sawTerminal) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event stream continued after a terminal event.', + { retryable: false }, + ); + } + + const { envelope, requestId } = parseConversationEnvelopeMetadata(page); + const data = conversationObject(envelope.data, 'data'); + conversationExactKeys(data, PAGE_DATA_KEYS, 'data'); + + const operation = parseConversationOperation(data.operation, 'data.operation'); + if (operation.id !== operationIdCanonical) { + throw new CaveConversationSchemaError('data.operation.id'); + } + const complete = conversationBoolean(data.complete, 'data.complete'); + const eventsValue = data.events; + if (!Array.isArray(eventsValue)) { + throw new CaveConversationSchemaError('data.events'); + } + let nextCursor: CaveConversationEventCursor | undefined; + if (data.cursor !== undefined) { + const cursorRecord = conversationObject(data.cursor, 'data.cursor'); + conversationExactKeys(cursorRecord, PAGE_CURSOR_KEYS, 'data.cursor'); + conversationBoolean(cursorRecord.hasMore, 'data.cursor.hasMore'); + if (cursorRecord.next !== undefined) { + nextCursor = conversationBoundedString( + cursorRecord.next, + 'data.cursor.next', + CAVE_CONTRACT_LIMITS.cursorCharacters, + { requireNonEmpty: true }, + ); + } + if (cursorRecord.current !== undefined) { + conversationBoundedString( + cursorRecord.current, + 'data.cursor.current', + CAVE_CONTRACT_LIMITS.cursorCharacters, + { requireNonEmpty: true }, + ); + } + } + + const events: CaveConversationEvent[] = []; + let lastAccepted: number | undefined; + let pageSawTerminal: boolean = sawTerminal; + + for (const [index, entry] of eventsValue.entries()) { + const event = parseConversationEvent( + entry, + operation.id, + `data.events[${index}]`, + ); + const { eventId } = event; + + // Exact duplicates at or below the accepted cursor are suppressed, + // never re-emitted and never treated as a gap. + if (eventId <= deliveredThrough) { + continue; + } + if (eventId > operation.latestEventId) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event page ran ahead of the operation record.', + { retryable: false }, + ); + } + if (lastAccepted === undefined) { + if ( + !(firstPage && options.resumeAfterOpaqueCursor === true) && + eventId !== deliveredThrough + 1 + ) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event page did not continue the event stream.', + { retryable: false }, + ); + } + } else if (eventId !== lastAccepted + 1) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event page was not contiguous.', + { retryable: false }, + ); + } + + // Any event in wire order after this page's terminal event is a + // protocol violation. + if (pageSawTerminal) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event page contained an event after the terminal event.', + { retryable: false }, + ); + } + + if (event.type === 'operation.stopping') { + if (sawStopping) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event page repeated stopping.', + { retryable: false }, + ); + } + sawStopping = true; + } + if (isTerminalEventType(event)) { + pageSawTerminal = true; + } + + lastAccepted = eventId; + events.push(event); + } + + if (complete && !CONVERSATION_TERMINAL_STATE_SET.has(operation.state)) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event page claimed completion for a non-terminal operation.', + { retryable: false }, + ); + } + const acceptedThrough = lastAccepted ?? deliveredThrough; + // A first page resuming behind an opaque cursor cannot verify where + // the accepted cursor sits (a terminal-cursor page arrives empty), so + // the completion bound is only enforceable once an event ID is known. + const opaqueResumeWithoutEvents = + firstPage && + options.resumeAfterOpaqueCursor === true && + lastAccepted === undefined; + if ( + complete && + !opaqueResumeWithoutEvents && + acceptedThrough !== operation.latestEventId + ) { + throw new CaveConversationResponseError( + 'invalid_response', + 'Conversation event page completed before the terminal event.', + { retryable: false }, + ); + } + + if (pageSawTerminal) { + sawTerminal = true; + } + firstPage = false; + + return { + operation, + events, + complete, + requestId, + ...(nextCursor === undefined ? {} : { nextCursor }), + }; + }, + }; +} + +/** + * The defined `reconcile_required` reasons, read from normalized error + * details without trusting the error shape. + */ +export function caveConversationReconcileReason( + error: unknown, +): CaveConversationReconcileReason | undefined { + if (typeof error !== 'object' || error === null) { + return undefined; + } + let code: unknown; + let details: unknown; + try { + code = Reflect.get(error, 'code'); + details = Reflect.get(error, 'details'); + } catch { + return undefined; + } + if (code !== 'reconcile_required' || typeof details !== 'object' || details === null) { + return undefined; + } + let reason: unknown; + try { + reason = Reflect.get(details, 'reason'); + } catch { + return undefined; + } + if (typeof reason !== 'string') { + return undefined; + } + return CAVE_CONVERSATION_RECONCILE_REASONS.includes( + reason as CaveConversationReconcileReason, + ) + ? (reason as CaveConversationReconcileReason) + : undefined; +} diff --git a/packages/cave/src/index.ts b/packages/cave/src/index.ts index a28557f..db391b3 100644 --- a/packages/cave/src/index.ts +++ b/packages/cave/src/index.ts @@ -67,6 +67,37 @@ export { CAVE_PAIRING_SCOPES, CAVE_PAIRING_STATUSES, } from './schemas.js'; +export { + CAVE_CONVERSATION_EVENT_TYPES, + CAVE_CONVERSATION_OPERATION_STATES, + CAVE_CONVERSATION_ORIGINATING_SCOPES, + CAVE_CONVERSATION_RECONCILE_REASONS, + CAVE_CONVERSATION_TERMINAL_STATES, + caveConversationReconcileReason, + createConversationEventTranslator, + validateConversationEventCursor, +} from './conversation-control.js'; +export type { + CaveConversationEvent, + CaveConversationEventBase, + CaveConversationEventPage, + CaveConversationEventPageRequest, + CaveConversationEventTranslator, + CaveConversationEventType, + CaveConversationOperation, + CaveConversationOperationId, + CaveConversationOperationKind, + CaveConversationOperationState, + CaveConversationOriginatingScope, + CaveConversationReconcileReason, + CaveConversationStreamOptions, + CaveConversationTranslatedPage, + CaveCreateConversationRequest, + CaveCreateConversationResult, + CaveRetryConversationTurnRequest, + CaveSendConversationMessageRequest, + CaveSendConversationMessageResult, +} from './conversation-control.js'; export type { CaveAuthorityBinding, CaveAuthorityBoundPairingExchange, diff --git a/packages/cave/src/transport.ts b/packages/cave/src/transport.ts index ec49b67..2d9790c 100644 --- a/packages/cave/src/transport.ts +++ b/packages/cave/src/transport.ts @@ -1,5 +1,11 @@ import type { OperationContext, PageOptions } from '@opencoven/sdk-core/browser'; +import type { + CaveConversationEventPageRequest, + CaveConversationOperationId, + CaveCreateConversationRequest, + CaveSendConversationMessageRequest, +} from './conversation-control.js'; import type { CaveAuthorityBinding, CaveAuthorityBoundPairingExchange, @@ -55,6 +61,36 @@ export interface CaveTransport { options: PageOptions, context?: OperationContext, ): Promise; + /** + * Conversational control is optional for every transport. The five Client + * v1 conversation-operation routes are not yet declared by the + * authoritative Cave contract fixture this SDK vendors, so no transport + * binds them today; the client reports a missing one as + * `unsupported_operation` rather than inventing a route. Results are + * `unknown` at this trust boundary and are validated by the client. + */ + createConversation?( + request: CaveCreateConversationRequest, + context?: OperationContext, + ): Promise; + sendConversationMessage?( + conversationId: string, + request: CaveSendConversationMessageRequest, + context?: OperationContext, + ): Promise; + getConversationOperation?( + operationId: CaveConversationOperationId, + context?: OperationContext, + ): Promise; + readConversationOperationEvents?( + operationId: CaveConversationOperationId, + page: CaveConversationEventPageRequest, + context?: OperationContext, + ): Promise; + stopConversationOperation?( + operationId: CaveConversationOperationId, + context?: OperationContext, + ): Promise; /** * The familiar operations are optional so that a transport written against * an older Cave still satisfies this interface. The client reports a missing diff --git a/tests/cave-conversation-control.spec.ts b/tests/cave-conversation-control.spec.ts new file mode 100644 index 0000000..33042a2 --- /dev/null +++ b/tests/cave-conversation-control.spec.ts @@ -0,0 +1,1248 @@ +import { + CaveClient, + caveConversationReconcileReason, + createConversationEventTranslator, + isCaveClientError, + validateConversationEventCursor, + type CaveTransport, +} from '@opencoven/cave-client'; +import { + parseConversationOperation, + parseCreateConversationRequest, + parseCreateConversationResult, + parseRetryConversationTurnRequest, + parseSendConversationMessageRequest, +} from '../packages/cave/src/conversation-control.js'; +import { describe, expect, test, vi } from 'vitest'; + +const OPERATION_ID = '018f4f1a-77c2-7a31-8a15-55a25aaba001'; +const OPERATION_ID_MIXED_CASE = '018F4F1A-77C2-7A31-8A15-55A25AABA001'; +const RETRY_OPERATION_ID = '018f4f1a-77c2-7a31-8a15-55a25aaba002'; +const RETRY_OPERATION_ID_MIXED_CASE = '018F4F1A-77C2-7A31-8A15-55A25AABA002'; +const CONVERSATION_ID = 'conversation.v1'; + +function envelope(data: unknown, overrides: Record = {}): unknown { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'conversation-messages', 'cursors'], + operations: [ + 'conversations.create', + 'messages.send', + 'operations.read', + 'operations.events', + 'operations.stop', + ], + requestId: 'req-1', + data, + ...overrides, + }; +} + +function operationRecord(overrides: Record = {}): Record { + return { + id: OPERATION_ID, + kind: 'messages.send', + state: 'running', + originatingScope: 'chat:write', + conversationId: CONVERSATION_ID, + inputTurnId: 'turn-1', + latestEventId: 3, + replayFloorEventId: 1, + createdAt: '2026-08-30T00:00:00.000Z', + updatedAt: '2026-08-30T00:00:01.000Z', + ...overrides, + }; +} + +function envelopeError( + code: string, + message: string, + details?: Record, +): unknown { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations'], + operations: ['messages.send'], + error: { + code, + message, + retryable: false, + ...(details === undefined ? {} : { details }), + }, + }; +} + +function transportWith(overrides: Record = {}): CaveTransport { + return { + health() { + throw new Error('health is not expected in this test'); + }, + ...overrides, + }; +} + +async function errorOf(run: () => Promise): Promise { + try { + await run(); + } catch (error) { + return error; + } + throw new Error('expected the call to reject'); +} + +describe('conversation mutation requests', () => { + test('rejects a malformed operation ID without echoing the untrusted value', async () => { + const client = new CaveClient({ + transport: { + health() { + throw new Error('unreachable'); + }, + } satisfies CaveTransport, + }); + + for (const malformed of [ + 'not-a-uuid', + '018f4f1a-77c2-7a31-8a15-55a25aaba00', // 35 characters + '018f4f1a77c27a318a1555a25aaba0011', // wrong shape + '018f4f1a-77c2-0a31-8a15-55a25aaba001', // version nibble out of range + ]) { + const error = await errorOf(() => + client.createConversation({ + operationId: malformed, + familiarId: 'familiar.v1', + }), + ); + expect(error).toBeInstanceOf(TypeError); + expect((error as Error).message).not.toContain(malformed); + expect(isCaveClientError(error)).toBe(false); + } + }); + + test('rejects unknown fields and non-canonical target IDs', async () => { + const client = new CaveClient({ + transport: { + health() { + throw new Error('unreachable'); + }, + } satisfies CaveTransport, + }); + + await expect( + client.createConversation({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + // @ts-expect-error probing an unknown field must not reach a transport + harness: 'bash', + }), + ).rejects.toThrowError(/unknown field/u); + + await expect( + client.createConversation({ operationId: OPERATION_ID, familiarId: '..' }), + ).rejects.toThrowError(/must not be a dot path segment/u); + }); + + test('requires exactly one of text or retryOfTurnId on a send', async () => { + const client = new CaveClient({ + transport: { + health() { + throw new Error('unreachable'); + }, + } satisfies CaveTransport, + }); + + await expect( + client.sendConversationMessage(CONVERSATION_ID, { + operationId: OPERATION_ID, + text: 'hello', + retryOfTurnId: 'turn-1', + } as never), + ).rejects.toThrowError(/exactly one of text or retryOfTurnId/u); + + await expect( + client.sendConversationMessage(CONVERSATION_ID, { + operationId: OPERATION_ID, + } as unknown as Parameters[1]), + ).rejects.toThrowError(/exactly one of text or retryOfTurnId/u); + + await expect( + client.sendConversationMessage(CONVERSATION_ID, { + operationId: OPERATION_ID, + text: ' ', + }), + ).rejects.toThrowError(/non-empty string/u); + }); + + test('normalizes the operation UUID and preserves text byte for byte', async () => { + let seenRequest: unknown; + const transport = transportWith({ + sendConversationMessage(_conversationId: string, request: unknown) { + seenRequest = request; + return envelope({ operation: operationRecord(), replayed: false }); + }, + }); + + const client = new CaveClient({ transport }); + const text = ' keep \nexact bytes\t'; + const result = await client.sendConversationMessage(CONVERSATION_ID, { + operationId: OPERATION_ID_MIXED_CASE, + text, + }); + + expect(seenRequest).toEqual({ operationId: OPERATION_ID, text }); + expect(result.operation.id).toBe(OPERATION_ID); + expect(result.operation.originatingScope).toBe('chat:write'); + expect(result.replayed).toBe(false); + }); + + test('retry carries a fresh operation UUID and explicit retryOfTurnId with no text', async () => { + let seenRequest: unknown; + const transport = transportWith({ + sendConversationMessage(_conversationId: string, request: unknown) { + seenRequest = request; + return envelope({ + operation: operationRecord({ + id: RETRY_OPERATION_ID, + retryOfTurnId: 'turn-7', + state: 'running', + }), + replayed: false, + }); + }, + }); + + const client = new CaveClient({ transport }); + const result = await client.retryConversationTurn( + CONVERSATION_ID, + { operationId: RETRY_OPERATION_ID_MIXED_CASE, retryOfTurnId: 'turn-7' }, + ); + + expect(seenRequest).toEqual({ + operationId: RETRY_OPERATION_ID, + retryOfTurnId: 'turn-7', + }); + expect(result.operation.id).toBe(RETRY_OPERATION_ID); + expect(result.operation.retryOfTurnId).toBe('turn-7'); + }); +}); + +describe('conversation create results and errors', () => { + test('returns the recorded create result with the normalized operation ID', async () => { + let seenRequest: unknown; + const transport = transportWith({ + createConversation(request: unknown) { + seenRequest = request; + return envelope({ + operationId: OPERATION_ID, + replayed: false, + conversation: { + id: CONVERSATION_ID, + familiarId: 'familiar.v1', + updatedAt: '2026-08-30T00:00:00.000Z', + }, + }); + }, + }); + + const client = new CaveClient({ transport }); + const result = await client.createConversation({ + operationId: OPERATION_ID_MIXED_CASE, + familiarId: 'familiar.v1', + }); + + expect(seenRequest).toEqual({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + }); + expect(result.operationId).toBe(OPERATION_ID); + expect(result.replayed).toBe(false); + expect(result.conversation.id).toBe(CONVERSATION_ID); + }); + + test('reports unsupported_operation with the operation ID when the transport cannot send', async () => { + // An older transport satisfies CaveTransport without the conversation + // methods; the client reports the missing capability instead of + // inventing a route. The accepted operation ID still rides the error. + const client = new CaveClient({ + transport: { + health() { + throw new Error('unreachable'); + }, + } satisfies CaveTransport, + }); + + const error = await errorOf(() => + client.createConversation({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + }), + ); + + expect(isCaveClientError(error)).toBe(true); + expect((error as { code: string }).code).toBe('unsupported_operation'); + expect((error as { operationId?: string }).operationId).toBe(OPERATION_ID); + }); + + test('attaches the operation ID to transport failures after acceptance and never retries', async () => { + const createConversation = vi.fn(() => { + throw new Error('connection reset while dispatching'); + }); + const client = new CaveClient({ + transport: transportWith({ createConversation }), + }); + + const error = await errorOf(() => + client.createConversation({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + }), + ); + + expect(isCaveClientError(error)).toBe(true); + expect((error as { operationId?: string }).operationId).toBe(OPERATION_ID); + expect(createConversation).toHaveBeenCalledTimes(1); + }); + + test('rejects a hostile create result as an invalid response with the operation ID', async () => { + const transport = transportWith({ + createConversation() { + return envelope({ + operationId: OPERATION_ID, + replayed: false, + conversation: { + id: CONVERSATION_ID, + familiarId: 'familiar.v1', + // updatedAt is required by the canonical conversation schema + }, + }); + }, + }); + const client = new CaveClient({ transport }); + + const error = await errorOf(() => + client.createConversation({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + }), + ); + + expect(isCaveClientError(error)).toBe(true); + expect((error as { code: string }).code).toBe('invalid_response'); + expect((error as { operationId?: string }).operationId).toBe(OPERATION_ID); + }); +}); + +describe('conversation operation read', () => { + test('returns the non-content operation record', async () => { + const transport = transportWith({ + getConversationOperation() { + return envelope({ + operation: operationRecord({ + state: 'running', + outputTurnId: undefined, + }), + }); + }, + }); + const client = new CaveClient({ transport }); + + const operation = await client.getConversationOperation(OPERATION_ID); + expect(operation.id).toBe(OPERATION_ID); + expect(operation.kind).toBe('messages.send'); + expect(operation.state).toBe('running'); + expect(operation.originatingScope).toBe('chat:write'); + expect(operation.conversationId).toBe(CONVERSATION_ID); + expect(operation.latestEventId).toBe(3); + expect(operation.replayFloorEventId).toBe(1); + }); + + test('rejects an operation record with an unknown field', async () => { + const transport = transportWith({ + getConversationOperation() { + return envelope({ + operation: operationRecord({ promptText: 'must never pass' }), + }); + }, + }); + const client = new CaveClient({ transport }); + + const error = await errorOf(() => client.getConversationOperation(OPERATION_ID)); + expect(isCaveClientError(error)).toBe(true); + expect((error as { code: string }).code).toBe('invalid_response'); + }); +}); + +describe('conversation stop', () => { + test('sends each explicit Stop exactly once and parses the resulting operation', async () => { + const stopConversationOperation = vi.fn(() => + envelope({ operation: operationRecord({ state: 'stopping' }) }), + ); + const client = new CaveClient({ + transport: transportWith({ stopConversationOperation }), + }); + + const operation = await client.stopConversationOperation(OPERATION_ID); + expect(operation.state).toBe('stopping'); + expect(stopConversationOperation).toHaveBeenCalledTimes(1); + }); + + test('does not retry Stop after an ambiguous transport completion', async () => { + const stopConversationOperation = vi.fn(() => { + throw new Error('connection reset after dispatch'); + }); + const client = new CaveClient({ + transport: transportWith({ stopConversationOperation }), + }); + + const error = await errorOf(() => client.stopConversationOperation(OPERATION_ID)); + expect(isCaveClientError(error)).toBe(true); + expect((error as { operationId?: string }).operationId).toBe(OPERATION_ID); + expect(stopConversationOperation).toHaveBeenCalledTimes(1); + }); +}); + +describe('reconciliation on replay gaps', () => { + test('surfaces reconcile_required with its reason as an instruction to reload', async () => { + const sendConversationMessage = vi.fn(() => + envelopeError('reconcile_required', 'Replay history is unavailable.', { + reason: 'replay_gap', + }), + ); + const client = new CaveClient({ + transport: transportWith({ sendConversationMessage }), + }); + + const error = await errorOf(() => + client.sendConversationMessage(CONVERSATION_ID, { + operationId: OPERATION_ID, + text: 'hello', + }), + ); + + expect(isCaveClientError(error)).toBe(true); + expect((error as { code: string }).code).toBe('reconcile_required'); + expect((error as { operationId?: string }).operationId).toBe(OPERATION_ID); + expect(caveConversationReconcileReason(error)).toBe('replay_gap'); + // An instruction to reload canonical history is never retried by the SDK. + expect(sendConversationMessage).toHaveBeenCalledTimes(1); + }); + + test('reads undefined for unrelated errors', () => { + const error = new Error('plain failure'); + expect(caveConversationReconcileReason(error)).toBeUndefined(); + expect(caveConversationReconcileReason(null)).toBeUndefined(); + expect( + caveConversationReconcileReason({ + code: 'reconcile_required', + details: { reason: 'something_else' }, + }), + ).toBeUndefined(); + }); +}); + +describe('operation record parsing', () => { + test('a read-only transport without conversation methods still satisfies CaveTransport', () => { + const transport: CaveTransport = { + health() { + throw new Error('unreachable'); + }, + }; + expect(() => new CaveClient({ transport })).not.toThrow(); + }); +}); + + +describe('conversation response envelope validation', () => { + const base = { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations'], + operations: ['messages.send'], + }; + + function clientReturning(response: unknown): CaveClient { + return new CaveClient({ + transport: transportWith({ + getConversationOperation() { + return response; + }, + }), + }); + } + + test('rejects incompatible api versions and newer client requirements', async () => { + const staleApi = clientReturning({ + ...base, + apiVersion: '2.0', + data: { operation: operationRecord() }, + }); + await expect(staleApi.getConversationOperation(OPERATION_ID)).rejects.toThrowError( + /invalid_response/u, + ); + + const newerClient = clientReturning({ + ...base, + minimumClientVersion: '99.0.0', + data: { operation: operationRecord() }, + }); + const error = await errorOf(() => + newerClient.getConversationOperation(OPERATION_ID), + ); + expect(isCaveClientError(error)).toBe(true); + expect((error as { code: string }).code).toBe('incompatible_version'); + }); + + test('rejects envelopes without exactly one of data or error', async () => { + await expect( + clientReturning(base).getConversationOperation(OPERATION_ID), + ).rejects.toThrowError(/invalid_response/u); + + await expect( + clientReturning({ + ...base, + data: { operation: operationRecord() }, + error: { code: 'conflict', message: 'both present', retryable: false }, + }).getConversationOperation(OPERATION_ID), + ).rejects.toThrowError(/invalid_response/u); + }); + + test('rejects malformed envelope declarations and overlong request ids', async () => { + await expect( + clientReturning({ + ...base, + capabilities: [], + data: { operation: operationRecord() }, + }).getConversationOperation(OPERATION_ID), + ).rejects.toThrowError(/invalid_response/u); + + await expect( + clientReturning({ + ...base, + requestId: 'x'.repeat(65), + data: { operation: operationRecord() }, + }).getConversationOperation(OPERATION_ID), + ).rejects.toThrowError(/invalid_response/u); + + // Additive unknown top-level envelope fields stay tolerated, matching the + // Client v1 additive-compatibility rule. + const additive = clientReturning({ + ...base, + unknownEnvelopeField: true, + data: { operation: operationRecord() }, + }); + const operation = await additive.getConversationOperation(OPERATION_ID); + expect(operation.kind).toBe('messages.send'); + }); + + test('rejects unknown error-envelope fields and unknown error codes', async () => { + await expect( + clientReturning({ + ...base, + error: { + code: 'conflict', + message: 'm', + retryable: false, + stack: 'private', + }, + }).getConversationOperation(OPERATION_ID), + ).rejects.toThrowError(/invalid_response/u); + + await expect( + clientReturning({ + ...base, + error: { code: 'not_an_error_code', message: 'm', retryable: false }, + }).getConversationOperation(OPERATION_ID), + ).rejects.toThrowError(/invalid_response/u); + }); + + test('caps error details at the contract limit', async () => { + const details: Record = {}; + for (let index = 0; index < 17; index += 1) { + details[`key${index}`] = 'value'; + } + await expect( + clientReturning({ + ...base, + error: { code: 'conflict', message: 'm', retryable: false, details }, + }).getConversationOperation(OPERATION_ID), + ).rejects.toThrowError(/invalid_response/u); + }); + + test('rejects result envelopes naming a different operation', async () => { + const sendClient = new CaveClient({ + transport: transportWith({ + sendConversationMessage() { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations'], + operations: ['messages.send'], + data: { + operation: operationRecord({ id: RETRY_OPERATION_ID }), + replayed: false, + }, + }; + }, + }), + }); + await expect( + sendClient.sendConversationMessage(CONVERSATION_ID, { + operationId: OPERATION_ID, + text: 'hello', + }), + ).rejects.toThrowError(/invalid_response/u); + + const createClient = new CaveClient({ + transport: transportWith({ + createConversation() { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations'], + operations: ['conversations.create'], + data: { + operationId: RETRY_OPERATION_ID, + replayed: false, + conversation: { + id: CONVERSATION_ID, + familiarId: 'familiar.v1', + updatedAt: 'x', + }, + }, + }; + }, + }), + }); + await expect( + createClient.createConversation({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + }), + ).rejects.toThrowError(/invalid_response/u); + }); +}); + +describe('conversation-control parser unit coverage', () => { + test('create requests accept the optional project ID', () => { + const parsed = parseCreateConversationRequest({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + projectId: 'project.v1', + }); + expect(parsed.projectId).toBe('project.v1'); + const minimal = parseCreateConversationRequest({ + operationId: OPERATION_ID, + familiarId: 'familiar.v1', + }); + expect(minimal.projectId).toBeUndefined(); + }); + + test('retry requests require retryOfTurnId and refuse extra fields', () => { + expect(() => parseRetryConversationTurnRequest({ operationId: OPERATION_ID })).toThrowError( + /requires retryOfTurnId/u, + ); + expect(() => + parseRetryConversationTurnRequest({ + operationId: OPERATION_ID, + retryOfTurnId: 'turn-1', + text: 'no replacement text', + }), + ).toThrowError(/unknown field/u); + expect(() => + parseRetryConversationTurnRequest({ + operationId: OPERATION_ID, + retryOfTurnId: 'turn-1', + }), + ).not.toThrow(); + }); + + test('rejects non-object and missing-key requests', () => { + expect(() => parseCreateConversationRequest(null)).toThrowError(/must be an object/u); + expect(() => parseCreateConversationRequest('nope')).toThrowError(/must be an object/u); + expect(() => parseCreateConversationRequest([])).toThrowError(/must be an object/u); + expect(() => + parseCreateConversationRequest({ operationId: OPERATION_ID }), + ).toThrowError(/requires familiarId/u); + expect(() => + parseSendConversationMessageRequest({ operationId: OPERATION_ID, text: 'x', extra: 1 }), + ).toThrowError(/unknown field/u); + }); + + test('rejects overlong and malformed event cursors', () => { + expect(() => + validateConversationEventCursor('x'.repeat(513), 'cursor'), + ).toThrowError(/at most 512 characters/u); + expect(() => validateConversationEventCursor('', 'cursor')).toThrowError( + /non-empty string/u, + ); + expect(() => validateConversationEventCursor(42, 'cursor')).toThrowError( + /non-empty string/u, + ); + expect(validateConversationEventCursor('a'.repeat(512), 'cursor')).toBe('a'.repeat(512)); + }); + + test('rejects non-monotonic translator commits', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + translator.commit(2); + expect(() => translator.commit(1)).toThrowError(/monotonically/u); + expect(() => translator.commit(0)).toThrowError(/monotonically/u); + expect(() => translator.commit(1.5)).toThrowError(/monotonically/u); + }); + + test('rejects create results naming a different operation ID', () => { + expect(() => + parseCreateConversationResult( + { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations'], + operations: ['conversations.create'], + data: { + operationId: OPERATION_ID_MIXED_CASE, + replayed: false, + conversation: { + id: CONVERSATION_ID, + familiarId: 'familiar.v1', + updatedAt: 'x', + }, + }, + }, + OPERATION_ID, + ), + ).not.toThrow(); + }); + + test('rejects operation records with contradictory bounds', () => { + expect(() => + parseConversationOperation( + operationRecord({ replayFloorEventId: 0 }), + 'data.operation', + ), + ).toThrowError(/data\.operation\.replayFloorEventId/u); + expect(() => + parseConversationOperation( + operationRecord({ replayFloorEventId: 5, latestEventId: 2 }), + 'data.operation', + ), + ).toThrowError(/replayFloorEventId/u); + expect(() => + parseConversationOperation( + operationRecord({ + state: 'completed', + outputTurnId: 'turn-2', + latestEventId: 0, + }), + 'data.operation', + ), + ).toThrowError(/data\.operation\.latestEventId/u); + expect(() => + parseConversationOperation( + operationRecord({ id: OPERATION_ID.toUpperCase() }), + 'data.operation', + ), + ).toThrowError(/data\.operation\.id was malformed/u); + }); + + test('parses failed and cancelled terminal events with their payloads', () => { + const failed = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }).translate( + envelope({ + operation: operationRecord({ + state: 'failed', + outputTurnId: 'turn-2', + failureCode: 'authority_restarted_during_execution', + latestEventId: 1, + }), + events: [ + { + type: 'operation.failed', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + outputTurnId: 'turn-2', + code: 'authority_restarted_during_execution', + }, + ], + complete: true, + }), + ); + expect(failed.events[0]).toMatchObject({ + type: 'operation.failed', + outputTurnId: 'turn-2', + code: 'authority_restarted_during_execution', + }); + }); + + test('parses stopping and accepted events with retry provenance', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + const translated = translator.translate( + envelope({ + operation: operationRecord({ latestEventId: 2 }), + events: [ + { + type: 'operation.accepted', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + conversationId: CONVERSATION_ID, + inputTurnId: 'turn-1', + retryOfTurnId: 'turn-7', + }, + { + type: 'operation.stopping', + operationId: OPERATION_ID, + eventId: 2, + cursor: 'cursor-2', + occurredAt: 'x', + }, + ], + complete: false, + }), + ); + expect(translated.events).toHaveLength(2); + expect(translated.events[0]).toMatchObject({ + type: 'operation.accepted', + retryOfTurnId: 'turn-7', + }); + }); +}); + +describe('reconcile reason hostile shapes', () => { + test('reads hostile error shapes without throwing', () => { + expect( + caveConversationReconcileReason({ + get code(): string { + throw new Error('private'); + }, + }), + ).toBeUndefined(); + expect( + caveConversationReconcileReason({ + code: 'reconcile_required', + details: { + get reason(): string { + throw new Error('private reason'); + }, + }, + }), + ).toBeUndefined(); + expect( + caveConversationReconcileReason({ code: 'reconcile_required', details: 42 }), + ).toBeUndefined(); + expect( + caveConversationReconcileReason({ code: 'reconcile_required', details: { reason: 7 } }), + ).toBeUndefined(); + expect( + caveConversationReconcileReason({ code: 'other', details: { reason: 'replay_gap' } }), + ).toBeUndefined(); + expect(caveConversationReconcileReason(undefined)).toBeUndefined(); + expect( + caveConversationReconcileReason({ + code: 'reconcile_required', + details: { reason: 'canonical_state_moved' }, + }), + ).toBe('canonical_state_moved'); + }); +}); + +describe('conversation stream translator extra coverage', () => { + function envelopeWith(data: unknown): unknown { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'cursors'], + operations: ['operations.events'], + data, + }; + } + + function acceptedEvent(): unknown { + return { + type: 'operation.accepted', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + conversationId: CONVERSATION_ID, + inputTurnId: 'turn-1', + }; + } + + function deltaFor(eventId: number): unknown { + return { + type: 'assistant.delta', + operationId: OPERATION_ID, + eventId, + cursor: `cursor-${eventId}`, + occurredAt: 'x', + text: 't', + }; + } + + test('accepts page cursors and exposes the next cursor', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + const translated = translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 4 }), + events: [deltaFor(2)], + complete: false, + cursor: { current: 'cursor-2', next: 'cursor-4', hasMore: true }, + }), + ); + expect(translated.nextCursor).toBe('cursor-4'); + }); + + test('rejects malformed page cursors and non-array events', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord(), + events: 'not-an-array', + complete: false, + }), + ), + ).toThrowError(/data\.events was malformed/u); + + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 4 }), + events: [], + complete: false, + cursor: { hasMore: false, current: 'x'.repeat(513) }, + }), + ), + ).toThrowError(/data\.cursor\.current was malformed/u); + }); + + test('rejects pages whose operation record names a different operation', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ id: RETRY_OPERATION_ID, latestEventId: 0 }), + events: [], + complete: false, + }), + ), + ).toThrowError(/data\.operation\.id/u); + }); + + test('refuses an event repeated within one page', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 3 }), + events: [acceptedEvent(), deltaFor(2), deltaFor(2), deltaFor(3)], + complete: false, + }), + ), + ).toThrowError(/was not contiguous/u); + }); + + test('resumes an empty page behind an opaque cursor and keeps polling', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + const translated = translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 4 }), + events: [], + complete: false, + }), + ); + expect(translated.events).toEqual([]); + expect(translated.complete).toBe(false); + }); +}); + +describe('conversation stream translator final branches', () => { + function envelopeWith(data: unknown): unknown { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'cursors'], + operations: ['operations.events'], + data, + }; + } + + function acceptedEvent(): unknown { + return { + type: 'operation.accepted', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + conversationId: CONVERSATION_ID, + inputTurnId: 'turn-1', + }; + } + + function deltaFor(eventId: number): unknown { + return { + type: 'assistant.delta', + operationId: OPERATION_ID, + eventId, + cursor: `cursor-${eventId}`, + occurredAt: 'x', + text: 't', + }; + } + + test('refuses a page after the stream already saw its terminal event', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + translator.translate( + envelopeWith({ + operation: operationRecord({ + state: 'completed', + outputTurnId: 'turn-2', + latestEventId: 2, + }), + events: [acceptedEvent(), { + type: 'operation.completed', + operationId: OPERATION_ID, + eventId: 2, + cursor: 'cursor-2', + occurredAt: 'x', + outputTurnId: 'turn-2', + }], + complete: true, + }), + ); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ + state: 'completed', + outputTurnId: 'turn-2', + latestEventId: 2, + }), + events: [], + complete: true, + }), + ), + ).toThrowError(/continued after a terminal event/u); + }); + + test('refuses events ahead of the operation record', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 4 }), + events: [deltaFor(9)], + complete: false, + }), + ), + ).toThrowError(/ran ahead of the operation record/u); + }); + + test('refuses a repeated stopping event', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 3 }), + events: [ + { + type: 'operation.stopping', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + }, + { + type: 'operation.stopping', + operationId: OPERATION_ID, + eventId: 2, + cursor: 'cursor-2', + occurredAt: 'x', + }, + ], + complete: false, + }), + ), + ).toThrowError(/repeated stopping/u); + }); +}); + + +describe('conversation event parser hostile payloads', () => { + function envelopeWith(data: unknown): unknown { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'cursors'], + operations: ['operations.events'], + data, + }; + } + + function deltaFor(eventId: number, extra: Record = {}): unknown { + return { + type: 'assistant.delta', + operationId: OPERATION_ID, + eventId, + cursor: `cursor-${eventId}`, + occurredAt: 'x', + text: 't', + ...extra, + }; + } + + test('rejects a foreign operation ID on an event', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 2 }), + events: [deltaFor(1, { operationId: RETRY_OPERATION_ID })], + complete: false, + }), + ), + ).toThrowError(/data\.events\[0\]\.operationId was malformed/u); + }); + + test('rejects an event id below one', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 2 }), + events: [deltaFor(0)], + complete: false, + }), + ), + ).toThrowError(/data\.events\[0\]\.eventId was malformed/u); + }); + + function deltaWithoutText(): unknown { + return { + type: 'assistant.delta', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + }; + } + + test('rejects a delta payload without text', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 2 }), + events: [deltaWithoutText()], + complete: false, + }), + ), + ).toThrowError(/data\.events\[0\]\.text was malformed/u); + }); + + test('rejects an event cursor that exceeds the contract limit', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 2 }), + events: [ + { + type: 'assistant.delta', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'x'.repeat(513), + occurredAt: 'x', + text: 't', + }, + ], + complete: false, + }), + ), + ).toThrowError(/data\.events\[0\]\.cursor was malformed/u); + }); + + test('rejects a stopping event repeating after a delivered stopping event', () => { + const first = createConversationEventTranslator(OPERATION_ID); + first.translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 2 }), + events: [ + { + type: 'operation.accepted', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + conversationId: CONVERSATION_ID, + inputTurnId: 'turn-1', + }, + { + type: 'operation.stopping', + operationId: OPERATION_ID, + eventId: 2, + cursor: 'cursor-2', + occurredAt: 'x', + }, + ], + complete: false, + }), + ); + expect(() => + createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }).translate( + envelopeWith({ + operation: operationRecord({ latestEventId: 3 }), + events: [ + { + type: 'operation.stopping', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: 'x', + }, + { + type: 'operation.stopping', + operationId: OPERATION_ID, + eventId: 2, + cursor: 'cursor-2', + occurredAt: 'x', + }, + ], + complete: false, + }), + ), + ).toThrowError(/repeated stopping/u); + }); +}); + +describe('conversation event translator identity', () => { + test('exposes its normalized operation ID and accepted cursor', () => { + const translator = createConversationEventTranslator(OPERATION_ID_MIXED_CASE); + expect(translator.operationId).toBe(OPERATION_ID); + expect(translator.deliveredThroughEventId).toBe(0); + }); +}); diff --git a/tests/cave-conversation-stream.spec.ts b/tests/cave-conversation-stream.spec.ts new file mode 100644 index 0000000..6fef532 --- /dev/null +++ b/tests/cave-conversation-stream.spec.ts @@ -0,0 +1,532 @@ +import { + CaveClient, + caveConversationReconcileReason, + createConversationEventTranslator, + isCaveClientError, + type CaveConversationEvent, + type CaveTransport, +} from '@opencoven/cave-client'; +import type { OperationContext } from '@opencoven/sdk-core'; +import { describe, expect, test, vi } from 'vitest'; + +const OPERATION_ID = '018f4f1a-77c2-7a31-8a15-55a25aaba001'; +const CONVERSATION_ID = 'conversation.v1'; + +function envelope(data: unknown): unknown { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'cursors'], + operations: ['operations.events'], + requestId: 'req-1', + data, + }; +} + +function envelopeError( + code: string, + message: string, + details?: Record, +): unknown { + return { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations'], + operations: ['operations.events'], + error: { + code, + message, + retryable: false, + ...(details === undefined ? {} : { details }), + }, + }; +} + +function operationRecord(overrides: Record = {}): Record { + return { + id: OPERATION_ID, + kind: 'messages.send', + state: 'running', + originatingScope: 'chat:write', + conversationId: CONVERSATION_ID, + inputTurnId: 'turn-1', + latestEventId: 4, + replayFloorEventId: 1, + createdAt: '2026-08-30T00:00:00.000Z', + updatedAt: '2026-08-30T00:00:01.000Z', + ...overrides, + }; +} + +function completedOperation(latestEventId: number): unknown { + return operationRecord({ + state: 'completed', + outputTurnId: 'turn-2', + latestEventId, + }); +} + +function acceptedEvent(): unknown { + return { + type: 'operation.accepted', + operationId: OPERATION_ID, + eventId: 1, + cursor: 'cursor-1', + occurredAt: '2026-08-30T00:00:01.000Z', + conversationId: CONVERSATION_ID, + inputTurnId: 'turn-1', + }; +} + +function deltaEvent(eventId: number, text: string): unknown { + return { + type: 'assistant.delta', + operationId: OPERATION_ID, + eventId, + cursor: `cursor-${eventId}`, + occurredAt: '2026-08-30T00:00:01.000Z', + text, + }; +} + +function completedEvent(eventId: number): unknown { + return { + type: 'operation.completed', + operationId: OPERATION_ID, + eventId, + cursor: `cursor-${eventId}`, + occurredAt: '2026-08-30T00:00:02.000Z', + outputTurnId: 'turn-2', + }; +} + +function runningPage(events: unknown[], latestEventId = 4, complete = false): unknown { + return envelope({ + operation: operationRecord({ latestEventId }), + events, + complete, + }); +} + +function completedPage(events: unknown[], latestEventId: number): unknown { + return envelope({ + operation: completedOperation(latestEventId), + events, + complete: true, + }); +} + +interface StreamHarness { + client: CaveClient; + reads: Array<{ cursor: string | undefined; deadline: number | undefined; aborted: boolean }>; + stop: ReturnType; + send: ReturnType; +} + +function streamHarness( + steps: Array<(context: OperationContext) => Promise>, +): StreamHarness { + const queue = [...steps]; + const reads: Array<{ + cursor: string | undefined; + deadline: number | undefined; + aborted: boolean; + }> = []; + const stop = vi.fn(() => { + throw new Error('stop must never be called by the stream'); + }); + const send = vi.fn(() => { + throw new Error('send must never be retried by the stream'); + }); + const readConversationOperationEvents = vi.fn( + async ( + _operationId: string, + pageRequest: { cursor?: string }, + context: OperationContext, + ): Promise => { + const step = queue.shift(); + if (step === undefined) { + throw new Error('unexpected extra event-page read'); + } + const record = { + cursor: pageRequest?.cursor, + deadline: context.deadline, + aborted: false, + }; + reads.push(record); + try { + return await step(context); + } catch (error) { + record.aborted = true; + throw error; + } + }, + ); + const transport = { + health() { + throw new Error('health is not expected in this test'); + }, + readConversationOperationEvents, + stopConversationOperation: stop, + sendConversationMessage: send, + } as unknown as CaveTransport; + return { client: new CaveClient({ transport }), reads, stop, send }; +} + +function delayThenPage(delayMs: number, page: unknown): () => Promise { + return async () => { + await new Promise((resolve) => setTimeout(resolve, delayMs)); + return page; + }; +} + +async function drain( + stream: AsyncGenerator, +): Promise { + const events: CaveConversationEvent[] = []; + for await (const event of stream) { + events.push(event); + } + return events; +} + +async function errorOf(run: () => Promise): Promise { + try { + await run(); + } catch (error) { + return error; + } + throw new Error('expected the call to reject'); +} + +describe('conversation event translator', () => { + test('translates one initial page in wire order with typed events', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + const translated = translator.translate( + runningPage([acceptedEvent(), deltaEvent(2, 'Hel'), deltaEvent(3, 'lo')]), + ); + + expect(translated.operation.id).toBe(OPERATION_ID); + expect(translated.events.map((event) => event.type)).toEqual([ + 'operation.accepted', + 'assistant.delta', + 'assistant.delta', + ]); + expect(translated.events[1]).toMatchObject({ type: 'assistant.delta', text: 'Hel' }); + expect(translated.complete).toBe(false); + }); + + test('suppresses an exact duplicate at or below the accepted cursor', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + const first = translator.translate( + runningPage([acceptedEvent(), deltaEvent(2, 'Hel'), deltaEvent(3, 'lo')]), + ); + expect(first.events.map((event) => event.eventId)).toEqual([1, 2, 3]); + for (const event of first.events) { + translator.commit(event.eventId); + } + expect(translator.deliveredThroughEventId).toBe(3); + + const second = translator.translate( + runningPage([deltaEvent(3, 're-delivered'), deltaEvent(4, '!')]), + ); + expect(second.events.map((event) => event.eventId)).toEqual([4]); + }); + + test('requires a fresh stream to begin at event 1', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + expect(() => + translator.translate(runningPage([deltaEvent(2, 'skipped one')])), + ).toThrowError(/did not continue the event stream/u); + }); + + test('refuses a forward gap within a page', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + expect(() => + translator.translate(runningPage([deltaEvent(2, 'a'), deltaEvent(4, 'gap')], 8)), + ).toThrowError(/was not contiguous/u); + }); + + test('refuses a forward gap across pages of one stream', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + translator.translate(runningPage([deltaEvent(2, 'baseline')], 8)); + translator.commit(2); + + expect(() => + translator.translate(runningPage([deltaEvent(4, 'gap')], 8)), + ).toThrowError(/did not continue the event stream/u); + }); + + test('refuses events after the terminal event in the same page', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + const hostile = { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'cursors'], + operations: ['operations.events'], + data: { + operation: operationRecord({ + state: 'completed', + outputTurnId: 'turn-2', + latestEventId: 3, + }), + events: [acceptedEvent(), completedEvent(2), deltaEvent(3, 'after terminal')], + complete: true, + }, + }; + expect(() => translator.translate(hostile)).toThrowError( + /after the terminal event/u, + ); + }); + + test('refuses a complete page for a non-terminal operation', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + expect(() => + translator.translate( + { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'cursors'], + operations: ['operations.events'], + data: { + operation: operationRecord({ latestEventId: 2 }), + events: [deltaEvent(1, 'a'), deltaEvent(2, 'b')], + complete: true, + }, + }, + ), + ).toThrowError(/non-terminal operation/u); + }); + + test('refuses a complete page that stops short of the terminal event', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + const shortPage = { + apiVersion: '1.0', + minimumClientVersion: '0.1.0', + capabilities: ['conversations', 'cursors'], + operations: ['operations.events'], + data: { + operation: completedOperation(4), + events: [deltaEvent(1, 'a'), deltaEvent(2, 'b'), completedEvent(3)], + complete: true, + }, + }; + expect(() => translator.translate(shortPage)).toThrowError( + /completed before the terminal event/u, + ); + }); + + test('reads a terminal-cursor page as complete with no events', () => { + const translator = createConversationEventTranslator(OPERATION_ID, { + resumeAfterOpaqueCursor: true, + }); + const final = translator.translate(completedPage([], 4)); + expect(final.complete).toBe(true); + expect(final.events).toEqual([]); + }); +}); + +describe('conversation route errors through the translator', () => { + test('preserves the reconcile reason from the error envelope', () => { + const translator = createConversationEventTranslator(OPERATION_ID); + let routeError: unknown; + try { + translator.translate( + envelopeError('reconcile_required', 'Replay history is unavailable.', { + reason: 'replay_gap', + }), + ); + } catch (error) { + routeError = error; + } + expect((routeError as { code?: string }).code).toBe('reconcile_required'); + expect((routeError as { details?: Record }).details?.reason).toBe( + 'replay_gap', + ); + }); +}); + +describe('streamConversationOperation', () => { + test('yields typed events in order and terminates at completion without Stop or resend', async () => { + const harness = streamHarness([ + () => Promise.resolve(runningPage([acceptedEvent(), deltaEvent(2, 'Hel'), deltaEvent(3, 'lo')], 4)), + () => Promise.resolve(completedPage([completedEvent(4)], 4)), + ]); + + const events = await drain(harness.client.streamConversationOperation(OPERATION_ID)); + + expect(events.map((event) => event.type)).toEqual([ + 'operation.accepted', + 'assistant.delta', + 'assistant.delta', + 'operation.completed', + ]); + expect(events[3]).toMatchObject({ type: 'operation.completed', outputTurnId: 'turn-2' }); + expect(harness.stop).not.toHaveBeenCalled(); + expect(harness.send).not.toHaveBeenCalled(); + }); + + test('suppresses a duplicate re-delivered within one stream', async () => { + const harness = streamHarness([ + () => Promise.resolve(runningPage([acceptedEvent(), deltaEvent(2, 'Hel'), deltaEvent(3, 'lo')], 4)), + () => Promise.resolve(runningPage([deltaEvent(3, 'duplicate'), deltaEvent(4, '!')], 4)), + () => Promise.resolve(completedPage([], 4)), + ]); + + const events = await drain(harness.client.streamConversationOperation(OPERATION_ID)); + expect(events.map((event) => event.eventId)).toEqual([1, 2, 3, 4]); + expect(harness.reads).toHaveLength(3); + }); + + test('terminates on an empty page when complete is true', async () => { + const harness = streamHarness([ + () => + Promise.resolve( + completedPage( + [acceptedEvent(), deltaEvent(2, 'a'), deltaEvent(3, 'b'), completedEvent(4)], + 4, + ), + ), + () => Promise.resolve(completedPage([], 4)), + ]); + + const events = await drain(harness.client.streamConversationOperation(OPERATION_ID)); + + expect(events).toHaveLength(4); + expect(harness.reads).toHaveLength(1); + }); + + test('keeps polling an empty page while the operation is not complete', async () => { + const harness = streamHarness([ + () => Promise.resolve(runningPage([], 4)), + () => Promise.resolve(runningPage([acceptedEvent()], 4)), + () => Promise.resolve(completedPage([completedEvent(2)], 2)), + ]); + + const events = await drain(harness.client.streamConversationOperation(OPERATION_ID)); + + expect(events.map((event) => event.eventId)).toEqual([1, 2]); + expect(harness.reads).toHaveLength(3); + }); +}); + + +describe('streamConversationOperation abort and resume', () => { + test('a caller abort closes the in-flight read and the generator without Stop or resend', async () => { + const controller = new AbortController(); + const harness = streamHarness([ + () => Promise.resolve(runningPage([acceptedEvent(), deltaEvent(2, 'Hel')], 4)), + (context: OperationContext) => + new Promise((_resolve, reject) => { + context.signal.addEventListener( + 'abort', + () => reject(new Error('read aborted')), + { once: true }, + ); + }), + ]); + + const iterator = harness.client + .streamConversationOperation(OPERATION_ID, { signal: controller.signal }) [Symbol.asyncIterator](); + + const first = await iterator.next(); + expect(first.value).toMatchObject({ eventId: 1, type: 'operation.accepted' }); + const second = await iterator.next(); + expect(second.value).toMatchObject({ eventId: 2 }); + + const third = iterator.next(); + await new Promise((resolve) => setTimeout(resolve, 5)); + expect(harness.reads).toHaveLength(2); + controller.abort(); + + const closed = await third; + expect(closed.done).toBe(true); + expect(harness.stop).not.toHaveBeenCalled(); + expect(harness.send).not.toHaveBeenCalled(); + }); +}); + + +describe('streamConversationOperation resume', () => { + test('emits only events after the supplied cursor on a resumed stream', async () => { + const controller = new AbortController(); + const harness = streamHarness([ + () => Promise.resolve(runningPage([acceptedEvent(), deltaEvent(2, 'Hel')], 4)), + () => Promise.resolve(runningPage([deltaEvent(3, 'lo'), deltaEvent(4, '!')], 4)), + () => Promise.resolve(completedPage([], 4)), + ]); + + const firstIterator = harness.client + .streamConversationOperation(OPERATION_ID, { signal: controller.signal }) [Symbol.asyncIterator](); + const firstResult = await firstIterator.next(); + expect(firstResult.value).toMatchObject({ eventId: 1 }); + const secondResult = await firstIterator.next(); + expect(secondResult.value).toMatchObject({ eventId: 2 }); + controller.abort(); + const firstClosed = await firstIterator.next(); + expect(firstClosed.done).toBe(true); + + const resumed = await drain( + harness.client.streamConversationOperation(OPERATION_ID, { + cursor: 'cursor-2', + }), + ); + expect(resumed.map((event) => event.eventId)).toEqual([3, 4]); + expect(harness.reads[1]?.cursor).toBe('cursor-2'); + expect(harness.reads[2]?.cursor).toBe('cursor-4'); + }); +}); + +describe('streamConversationOperation budget', () => { + test('each long poll receives only the remaining budget', async () => { + const harness = streamHarness([ + delayThenPage(60, runningPage([acceptedEvent()], 4)), + delayThenPage(60, runningPage([deltaEvent(2, 'a')], 4)), + delayThenPage(500, runningPage([deltaEvent(3, 'b')], 4)), + ]); + + const error = await errorOf(async () => { + for await (const event of harness.client.streamConversationOperation(OPERATION_ID, { + timeoutMs: 150, + })) { + void event; + } + }); + + expect(isCaveClientError(error)).toBe(true); + expect((error as { code: string }).code).toBe('timeout'); + expect((error as { operationId?: string }).operationId).toBe(OPERATION_ID); + // The stream died on budget exhaustion: three polls were attempted and + // the third was cut off by the shared deadline. + expect(harness.reads).toHaveLength(3); + expect(harness.reads[2]?.deadline).toBeDefined(); + }); +}); + +describe('streamConversationOperation route errors', () => { + test('surfaces reconcile_required from the route as an instruction to reload', async () => { + const harness = streamHarness([ + () => + Promise.resolve( + envelopeError('reconcile_required', 'Replay history is unavailable.', { + reason: 'replay_gap', + }), + ), + ]); + + const error = await errorOf(async () => { + for await (const event of harness.client.streamConversationOperation(OPERATION_ID)) { + expect(event).toBeDefined(); + } + }); + + expect(isCaveClientError(error)).toBe(true); + expect((error as { code: string }).code).toBe('reconcile_required'); + expect((error as { operationId?: string }).operationId).toBe(OPERATION_ID); + expect(caveConversationReconcileReason(error)).toBe('replay_gap'); + }); +}); diff --git a/tests/public-contract.spec.ts b/tests/public-contract.spec.ts index ea2a13f..dae6034 100644 --- a/tests/public-contract.spec.ts +++ b/tests/public-contract.spec.ts @@ -434,6 +434,11 @@ describe('public package entry points', () => { expect(exportedKeys(cave)).toEqual([ 'CAVE_ANALYTICS_WINDOWS', 'CAVE_CLIENT_VERSION', + 'CAVE_CONVERSATION_EVENT_TYPES', + 'CAVE_CONVERSATION_OPERATION_STATES', + 'CAVE_CONVERSATION_ORIGINATING_SCOPES', + 'CAVE_CONVERSATION_RECONCILE_REASONS', + 'CAVE_CONVERSATION_TERMINAL_STATES', 'CAVE_FAMILIAR_PROPERTIES', 'CAVE_PAIRING_SCOPES', 'CAVE_PAIRING_STATUSES', @@ -441,7 +446,9 @@ describe('public package entry points', () => { 'CaveClientError', 'CaveDiscoveryError', 'CavePairingSession', + 'caveConversationReconcileReason', 'createCaveClient', + 'createConversationEventTranslator', 'createDiscoveredCaveClient', 'createManagedCaveClient', 'digestCaveContractFixture', @@ -451,6 +458,7 @@ describe('public package entry points', () => { 'normalizeCaveError', 'parseCaveContractFixture', 'parseVerifiedCaveContractFixture', + 'validateConversationEventCursor', 'verifyCaveContractFixtureDigest', ]); expect(exportedKeys(coven)).toEqual([