import { getRetainedProviderCommand } from "@opengeni/db/retained-provider-commands"; import { scheduledSessionIds } from "@opengeni/db"; import { withSiteSessionOrigin } from "@opengeni/core"; import { resolveSiteSessionOrigin } from "../site-session-origin"; import { AcknowledgeStreamRequest, ApplySessionGoalRevisionRequest, ListSessionGoalRevisionsQuery, RejectSessionGoalRevisionRequest, RollbackSessionGoalRevisionRequest, ActivateCodexRealtimeConnectionRequest, AttachViewerRequest, BeginSessionRealtimeRequest, ClearSessionContextRequest, CodexRealtimeWebrtcRequest, GatewayRealtimeConnectRequest, ClientSessionEvent, CompactSessionContextRequest, CreateSessionRequest, DeleteSessionQueueItemRequest, EditSessionQueueItemRequest, EndSessionRealtimeRequest, FsDeleteRequest, FsListBatchRequest, FsListRequest, FsMkdirRequest, FsMoveRequest, FsReadRequest, FsWriteRequest, ForkSessionRequest, HumanInputRequestStatus, GitDiffRequest, GitReadBatchRequest, GitLogRequest, GitShowRequest, GitStatusRequest, MoveSessionQueueItemRequest, PtyCloseRequest, PtyOpenRequest, PtyResizeRequest, PtyWriteRequest, PublishSandboxFileArtifactRequest, RenewSessionRealtimeRequest, SyncSessionRealtimeLedgerRequest, SessionControlRequest, SESSION_EVENT_RAW_DELTA_TYPES, SessionEventPayloadMode, SessionEventReadDirection, SessionEventReadMode, SessionEventLatestClass, SessionEventResultMode, SessionScopeSubjectId, SessionEventSemanticClass, SessionEventType, SessionMcpServerId, SaveNewSessionDraftRequest, compactSessionEventResult, sessionEventLatestClassToSemanticClass, SaveComposerDraftRequest, SubmitComposerDraftRequest, SteerSessionQueueItemRequest, SteerSessionMessageRequest, TerminalExecRequest, UpdateSessionChannelRequest, UpdateSessionAttentionRequest, UpdateSessionArchiveRequest, UpdateSessionPinRequest, UpdateSessionGoalRequest, UpdateSessionMcpApprovalPolicyRequest, UpdateSessionRequest, UpdateSessionVariableSetsRequest, UpdateSessionVisibilityRequest, UpdateSessionToolPolicyRequest, ViewerHeartbeatRequest, WORKSPACE_CONTROL_ACTOR_MAX_BYTES, WORK_CLAIM_CANONICAL_KEY_MAX_BYTES, WORK_CLAIM_DISCOVERY_LIMIT, WORK_CLAIM_NAMESPACE_MAX_BYTES, WORK_DISCOVERY_QUERY_MAX_CHARS, WORK_DISCOVERY_RECENT_HOURS_MAX, WorkClaimSubjectFilter as WorkClaimSubjectFilterSchema, WorkClaimSubjectType, normalizeWorkClaimCanonicalKey, normalizeWorkClaimNamespace, workspaceControlUtf8Bytes, type AccessGrant, type AttachViewerResponse, type SandboxBackend, type LineageNode, type Session, type SessionStatus, type WorkClaimSubjectFilter, type SessionGoalRevision, type AgentTopologyPageResponse, type ErrorCode, type SessionAuthorizationOperation, type SessionQueueSnapshot, type TerminalPtyExitedPayload, type TerminalPtyOutputDeltaPayload, type TerminalPtyStartedPayload, type VariableSet, } from "@opengeni/contracts"; import { streamTokenDegraded } from "@opengeni/config"; import { recordAuditEvent, acceptSessionApprovalDecision, acceptSessionHumanInputResponse, clearSessionGoal, clearSessionContext, getOpenPtySession, getEnrollment, getLiveEnrollmentConnection, getRetainedProcess, getSandbox, getSession, readActiveSandbox, deleteSessionTreeIfQuiescent, getSessionEvent, getSessionForSubject, getSessionGoal, getLatestSessionModelContext, getSessionHumanInputRequest, getSessionGoalWithContinuation, getSessionGoalRevision, getSessionQueueSnapshot, getStreamAcknowledgment, insertPtySession, listSessionEventPage, listSessionHumanInputRequests, listSessionGoalRevisionPage, listSessionGoalRevisions, rejectSessionGoalRevisionWithEvent, listSessionIdsInGroup, listSessionDiscoverySummaries, listSessionDiscoveryAncestorPaths, listSessionsForSubject, getLatestStartedSessionTurn, listSessionTurns, projectEffectiveControlForRelatedAccess, projectSessionForRelatedAccess, recordStreamAcknowledgment, requestSessionCompaction, switchSessionCodexAccount, setSessionChannel, updateSessionVariableSets, ChannelNotFoundError, setSessionAttention, setSessionArchive, setSessionPin, SessionAttentionVersionConflictError, SessionArchiveVersionConflictError, SessionPinVersionConflictError, SessionPinAccessError, SessionListAccessError, SessionListCursorError, SessionListCursorExpiredError, decodeSessionListCursor, revokeViewer, setSessionGoalStatusWithEvent, upsertSessionGoalWithEvent, updatePtySessionActivity, QueueCommandConflictError, beginSessionRealtimeInTransaction, activateSessionRealtimeConnectionInTransaction, claimSessionRealtimeConnectionInTransaction, completeSessionRealtimeConnectionInTransaction, endSessionRealtimeInTransaction, failSessionRealtimeConnectionInTransaction, NewSessionDraftConflictError, SessionCommandIdempotencyError, SessionControlConflictError, SessionRealtimeConflictError, SessionToolPolicyVersionConflictError, SessionContextBusyError, SessionVariableSetSelectionUnavailableError, workspaceControlRequestLockTimeoutMs, SessionTenancyAccessError, SessionTenancyConflictError, SessionTenancyInvalidRequestError, SessionTenancyNotActivatedError, HumanInputResponseValidationError, SkillHumanResponseError, latestWorkspaceCapture, sessionLatestWorkspaceCapture, renewSessionRealtimeInTransaction, syncSessionRealtimeLedgerInTransaction, withWorkspaceSessionActivityRls, withWorkspaceRls, workspaceCaptureAtRevision, type AppendEventInput, type SandboxOpenPtySessionRow, type SandboxPtyProcessIdentity, type SandboxRetainedProcess, type Database, type SessionDiscoveryCursor, type SessionDiscoveryOrderBy, type SessionDiscoveryAncestor, } from "@opengeni/db"; import { backgroundCommandActivityForSessions, listSessionBackgroundCommands, requestSessionBackgroundCommandCancellation, } from "@opengeni/db/session-background-commands"; import { appendAndPublishEvents, boundSessionEventHttpPage, coalesceSessionEventDeltasWithCoverage, publishDurableSessionEvents, } from "@opengeni/events"; import { createGatewayRealtimeConnectionSecret, GatewayRealtimeBrokerError, } from "../gateway-realtime"; import { createXaiRealtimeConnectionSecret, XaiRealtimeBrokerError } from "../xai-realtime"; import { prepareExternalLinkTurnAdmission, externalContinuationCommitAuthorizer, } from "@opengeni/core"; import { z, ZodError } from "zod"; import { runConcurrentChannelAReads, withChannelA, withChannelARead, type ChannelAContext, type ChannelAHandle, type ChannelAOperation, } from "../sandbox/channel-a"; import { NatsControlRpc, negotiateCapabilities, negotiateSelfhostedCapabilities, resolveConnectedMachineWorkspaceRoot, selectBackend, SelfhostedSession, } from "@opengeni/runtime/sandbox"; import type { Context, Hono, MiddlewareHandler } from "hono"; import { HTTPException } from "hono/http-exception"; import type { ContentfulStatusCode } from "hono/utils/http-status"; import { getManagedAuthRequestActorEpoch, hasPermission, requireAccessGrant, requireAccessGrantAuthorization, requireFreshAccessGrant, rotateSessionMcpCredentialsForRequest, hasVerifiedOwningUserAuthorization, requirePermission, requireSessionAuthorization, requireSessionAuthorizationListScope, resolveWorkspaceCatalogSettings, withResolvedSessionAuthorization, SESSION_AUTHORIZATION_DEFAULT_REAUTHORIZE_MS, SessionAuthorizationDeniedError, SessionAuthorizationUnavailableError, SessionTenancyManagedHumanRequiredError, SessionTenancyPersistenceOutcomeUnknownError, type ResolvedSessionAuthorization, } from "@opengeni/core"; import type { ApiRouteDeps } from "@opengeni/core"; import { attachViewer, detachViewer, heartbeatViewer, mintDesktopStream, mintTerminalStream, readGroupLease, resolveActiveDesktopTransport, viewerHeartbeatIntervalMs, type DesktopStreamMint, type TerminalStreamMint, type ViewerServices, } from "../sandbox/viewer"; import { buildSessionCodexRealtimeBroker, CodexRealtimeBrokerError } from "../codex-realtime"; import { acceptSessionUserMessage, controlHumanSessionWorkstream, createSessionForRequest, deleteHumanQueuePrompt, editHumanQueuePrompt, getActorNewSessionDraft, getManagedHumanSessionCreateCapabilities, getHumanComposerDraft, forkManagedHumanSession, moveHumanQueuePrompt, readSessionLineage, saveHumanComposerDraft, saveActorNewSessionDraft, SessionSpawnDeniedError, sessionSpawnDenialEnvelope, steerHumanQueuePrompt, submitComposerDraftForRequest, updateSessionMcpApprovalPolicy, updateManagedHumanSessionVisibility, updateSessionToolPolicy, updateSessionTitle, workflowIdForSession, sessionWithEffectiveToolPolicy, workspaceSessionToolPolicyDefaultServerIds, workspaceSessionToolPolicyServerIds, relayConfigFromSettings, validateVariableSetAttachment, } from "@opengeni/core"; import { assertSessionExists, boundedLimit } from "../http/common"; import { browserSseDeliveryOptions, sseSessionStream } from "../http/sse"; import { serveWorkspaceCapture, serveWorkspaceCaptureFile, WorkspaceCaptureManifestCache, } from "./workspace-capture"; import { publishSandboxFileArtifact } from "../sandbox-file-artifacts"; import { ApiHttpError } from "../http/api-error"; import { observeWorkDiscovery, summarizeWorkDiscoveryRows } from "../work-discovery-observability"; type SessionRouteDeps = ApiRouteDeps & Pick; const VIEWER_LIFECYCLE_PERMISSIONS = ["stream:view", "terminal:attach", "files:write"] as const; /** * Bounded responder kind recorded on a child's requires-action resolution * notice for its parent: a key principal is `api`, a signed agent attempt is * `agent_attempt`, and every other HTTP caller is a human. Never a subject id. */ function childRequiresActionRespondedByKindForGrant( grant: Pick, ): "human" | "api" | "agent_attempt" | "system" { if (grant.principalKind === "api_key" || grant.principalKind === "configured_key") return "api"; if (grant.principalKind === "agent_attempt") return "agent_attempt"; if (grant.principalKind === "service") return "system"; return "human"; } function requireViewerLifecyclePermission(grant: AccessGrant): void { if ( VIEWER_LIFECYCLE_PERMISSIONS.some((permission) => hasPermission(grant.permissions, permission)) ) { return; } throw new HTTPException(403, { message: `missing permission: ${VIEWER_LIFECYCLE_PERMISSIONS.join(" or ")}`, }); } export function registerSessionRoutes(app: Hono, deps: SessionRouteDeps): void { const { settings, db, bus, workflowClient, objectStorage } = deps; const channelAServices = { db, settings, bus, objectStorage, observability: deps.observability, }; const workspaceCaptureManifestCache = new WorkspaceCaptureManifestCache(); const ptyIdentity = (pty: SandboxOpenPtySessionRow): SandboxPtyProcessIdentity => ({ leaseId: pty.leaseId, sandboxGroupId: pty.sandboxGroupId, retainedProcessId: pty.retainedProcessId, openAdmissionId: pty.openAdmissionId, execSessionId: pty.execSessionId, leaseEpoch: pty.leaseEpoch, providerBackend: pty.providerBackend, providerInstanceId: pty.providerInstanceId, routeKind: pty.routeKind, routeTargetId: pty.routeTargetId, routeEpoch: pty.routeEpoch, }); const adoptPtyProcess = async ( ctx: ChannelAContext, handle: ChannelAHandle, pty: SandboxOpenPtySessionRow, ): Promise => { if (!handle.lease) { throw new HTTPException(409, { message: "durable interactive terminals require a session-home provider lease", }); } const process = await getRetainedProcess(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, processId: pty.retainedProcessId, }); if ( !process || process.state !== "active" || process.ownerActorKind !== "direct" || process.accountId !== ctx.accountId || process.leaseId !== pty.leaseId || process.sandboxGroupId !== pty.sandboxGroupId || process.parentAdmissionId !== pty.openAdmissionId || process.leaseEpoch !== pty.leaseEpoch || process.providerBackend !== pty.providerBackend || process.providerInstanceId !== pty.providerInstanceId || process.routeKind !== pty.routeKind || process.routeTargetId !== pty.routeTargetId || process.routeEpoch !== pty.routeEpoch || process.providerSessionId !== pty.execSessionId || // Only a persistable home backend can currently be reconstructed by an // API request without consulting the mutable active pointer. process.routeTargetId !== null || handle.lease.id !== process.leaseId || handle.lease.sandboxGroupId !== process.sandboxGroupId || handle.lease.leaseEpoch !== process.leaseEpoch || handle.lease.backend !== process.providerBackend || handle.lease.instanceId !== process.providerInstanceId ) { throw new HTTPException(409, { message: "pty retained-process identity is stale; reopen the terminal", }); } const providerCommand = await getRetainedProviderCommand(db, { accountId: ctx.accountId, workspaceId: ctx.workspaceId, sessionId: ctx.session.id, processId: process.id, }); handle.routingSession.adoptRetainedProcess({ process: { id: process.id, providerSessionId: process.providerSessionId, ...(providerCommand ? { providerCommand } : {}), }, backend: { sandboxId: null, leaseEpoch: process.leaseEpoch, providerInstanceId: process.providerInstanceId, activeEpoch: process.routeEpoch, }, }); return process; }; const emitPtyExited = async ( ctx: ChannelAContext, ptyId: string, process: SandboxRetainedProcess, ): Promise => { const exited: TerminalPtyExitedPayload = { ptyId, exitCode: process.exitCode, reason: process.state === "exited" ? "exit" : "lost", }; await appendAndPublishEvents(db, bus, ctx.workspaceId, ctx.session.id, [ { type: "terminal.pty.exited", payload: exited }, ]); }; const drainOpenedPty = async (handle: ChannelAHandle, execSessionId: number): Promise => { let chars = "\u0004"; while (handle.routingSession.hasRetainedProcess(execSessionId)) { await handle.routingSession.writeStdinForProcessControl({ sessionId: execSessionId, chars, yieldTimeMs: 250, maxOutputTokens: 128, }); chars = ""; } }; const failPtyPersistenceAndDrain = (persistenceError: unknown, drainError: unknown): never => { throw new AggregateError( [persistenceError, drainError], "PTY persistence failed and the exact opened process could not be drained", { cause: drainError }, ); }; const requestSessionAuthorization = new WeakMap(); const relatedSessionAccessFor = (c: Context): "target" | "root" => requestSessionAuthorization.get(c.req.raw)?.relatedSessionAccess ?? "root"; const projectQueueSnapshot = ( snapshot: SessionQueueSnapshot, sessionId: string, access: "target" | "root", ): SessionQueueSnapshot => ({ ...snapshot, effectiveControl: projectEffectiveControlForRelatedAccess( snapshot.effectiveControl, sessionId, access, ), }); // Every deployment has one fail-closed authorization seam for every HTTP // session surface. The core boundary always enforces durable OpenGeni-owned // private-session rules; an embedding host port can add narrower policy. // Register it before the routes so a newly added path cannot accidentally // inherit workspace access without an explicit operation classification. The // long-lived event stream performs its own initial check and bounded // reauthorization below. const authorizeSessionHttp: MiddlewareHandler = async (c, next) => { const workspaceId = c.req.param("workspaceId") ?? ""; const sessionId = c.req.param("sessionId") ?? ""; // Reject malformed route identifiers before the authorization resolver // reaches UUID-typed persistence queries. Besides avoiding a needless DB // round trip, this preserves the session surface's non-enumerating 404 // contract instead of leaking a driver-level 500. if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const operation = sessionAuthorizationOperationForHttp( c.req.method, new URL(c.req.url).pathname, sessionId, ); if (operation === "session.stream.read") { await next(); return; } if (!operation) { throw sessionAuthorizationHttpError(new SessionAuthorizationUnavailableError()); } if (operation === "session.visibility.write" || operation === "session.fork.create") { // These product mutations perform their target-free managed-cookie, // permission, and activation gates inside core before resolving the // target exactly once. Visibility changes remain owner-only; a shared // source may be forked by any currently authorized workspace member. // Running the generic middleware here would both create a pre-gate // existence oracle and double-call the host. await next(); return; } const grant = await requireAccessGrant(c, deps, workspaceId); try { const authorization = await requireSessionAuthorization(deps, grant, { sessionId, operation, surface: "http", }); if (authorization) requestSessionAuthorization.set(c.req.raw, authorization); if (authorization) { await withResolvedSessionAuthorization(authorization, next); return; } } catch (error) { throw sessionAuthorizationHttpError(error); } await next(); }; app.use("/v1/workspaces/:workspaceId/sessions/:sessionId/*", authorizeSessionHttp); const viewerServices: ViewerServices = { db, settings, bus, objectStorage, ...(deps.establishSandboxSession ? { establishSandboxSession: deps.establishSandboxSession } : {}), }; app.get("/v1/workspaces/:workspaceId/session-tenancy/capabilities", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:create", ); return c.json(await getManagedHumanSessionCreateCapabilities(deps, authorization, workspaceId)); }); app.post("/v1/workspaces/:workspaceId/sessions", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:create", ); const grant = authorization.grant; let payload: unknown; try { payload = await c.req.json(); } catch { return c.json( { code: "INVALID_SESSION_CREATE_REQUEST", message: "Invalid session create request: request body must contain valid JSON", }, 422, ); } let session: Session; try { CreateSessionRequest.parse(payload); const origin = await resolveSiteSessionOrigin( db, workspaceId, c.req.header("x-opengeni-site-id"), c.req.header("x-opengeni-site-version"), ); const create = () => createSessionForRequest(deps, grant, workspaceId, payload, authorization); session = await (origin ? withSiteSessionOrigin(origin, create) : create()); } catch (error) { return sessionCreateErrorResponse(c, error); } // Creation has committed by this point. Keep response projection outside // the create-rejection boundary so a post-commit policy read cannot be // misreported as though the session itself was rejected. return c.json(await withEffectivePolicy(deps, workspaceId, grant.subjectId, session), 202); }); app.get("/v1/workspaces/:workspaceId/new-session-draft", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:read", ); const grant = authorization.grant; const catalog = await resolveWorkspaceCatalogSettings(db, settings, { accountId: grant.accountId, workspaceId, }); return c.json( await getActorNewSessionDraft( { settings: catalog.settings, db }, grant, workspaceId, authorization, ), ); }); app.put("/v1/workspaces/:workspaceId/new-session-draft", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:create", ); const grant = authorization.grant; let payload: unknown; try { payload = await c.req.json(); } catch { return c.json( { code: "INVALID_NEW_SESSION_DRAFT_REQUEST", message: "Invalid new-session draft request: request body must contain valid JSON", }, 422, ); } try { SaveNewSessionDraftRequest.parse(payload); const catalog = await resolveWorkspaceCatalogSettings(db, settings, { accountId: grant.accountId, workspaceId, }); return c.json( await saveActorNewSessionDraft( { settings: catalog.settings, db, objectStorage }, grant, workspaceId, payload, authorization.canonicalManagedHumanSession, authorization, ), ); } catch (error) { if (error instanceof NewSessionDraftConflictError) { return c.json( { code: "NEW_SESSION_DRAFT_CONFLICT", message: error.message, currentRevision: error.currentRevision, }, 409, ); } if (error instanceof ZodError) { return c.json( { code: "INVALID_NEW_SESSION_DRAFT_REQUEST", message: `Invalid new-session draft request: ${zodErrorFields(error)} failed schema validation`, }, 422, ); } throw error; } }); app.get("/v1/workspaces/:workspaceId/sessions", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:read", ); const grant = authorization.grant; let authorizationScope; try { authorizationScope = await requireSessionAuthorizationListScope(deps, grant, "http"); } catch (error) { throw sessionAuthorizationHttpError(error); } const pageView = c.req.query("view") === "page"; const query = sessionListQuery(c.req.query(), pageView); let page: Awaited>; try { page = await listSessionsForSubject(db, workspaceId, { subjectId: grant.subjectId, limit: boundedLimit(query.limit), materializeSnapshot: pageView, ...(query.cursor ? { cursor: query.cursor } : {}), ...(query.search ? { search: query.search } : {}), ...(query.pinsOnly ? { pinsOnly: true } : {}), ...(query.archivedOnly ? { archivedOnly: true } : {}), ...(query.parentSessionId !== undefined ? { parentSessionId: query.parentSessionId } : {}), ...(query.channelId !== undefined ? { channelId: query.channelId } : {}), ...(query.originSiteId ? { originSiteId: query.originSiteId } : {}), ...(query.createdBy ? { createdBy: query.createdBy } : {}), ...(query.updatedFrom ? { updatedFrom: query.updatedFrom } : {}), ...(query.updatedBefore ? { updatedBefore: query.updatedBefore } : {}), ...(query.createdFrom ? { createdFrom: query.createdFrom } : {}), ...(query.createdBefore ? { createdBefore: query.createdBefore } : {}), ...(query.scopeSubjectId ? { scopeSubjectId: query.scopeSubjectId } : {}), ...(authorizationScope ? { authorizationScope } : {}), // A managed human's own personal workspace has no membership row, so // the list's removal fence must fall back to the organization-membership // pointer, only with verified native or external owning-user provenance. personalWorkspaceOwnerException: hasVerifiedOwningUserAuthorization(authorization), }); } catch (error) { if (error instanceof SessionListAccessError) { throw new HTTPException(403, { message: error.message }); } if (error instanceof SessionListCursorExpiredError) { // The caller's short-lived snapshot is no longer usable. Keep this // distinct from auth, network, and validation failures so clients can // rebase a retained continuation exactly once instead of retrying the // expired cursor forever. throw new HTTPException(410, { message: error.message }); } if (error instanceof SessionListCursorError) { throw new HTTPException(400, { message: error.message }); } throw error; } // The page body carries this fact directly. Preserve the historical array // body for older clients while still making its older-pin omission visible // to raw HTTP consumers without changing that response shape. c.header("x-opengeni-pinned-truncated", page.pinnedTruncated === true ? "true" : "false"); const policy = await loadEffectivePolicyContext(deps, workspaceId, grant.subjectId); const commandActivity = await backgroundCommandActivityForSessions(db, { accountId: grant.accountId, workspaceId, sessionIds: [...page.pinned, ...page.sessions].map((session) => session.id), }); const scheduleTargets = hasPermission(grant.permissions, "scheduled_tasks:run") && hasPermission(grant.permissions, "sessions:control") ? await scheduledSessionIds( db, workspaceId, [...page.pinned, ...page.sessions].map((session) => session.id), ) : new Set(); const decorate = (session: Session): Session => { const activity = commandActivity.get(session.id); return sessionWithEffectiveToolPolicy( { ...session, hasSchedules: scheduleTargets.has(session.id), ...(activity ? { backgroundCommandActivity: activity } : {}), }, policy.workspaceServerIds, policy.workspaceDefaultServerIds, ); }; if (pageView) { return c.json({ ...page, ...(query.hasPageFilters ? { filtersApplied: true as const } : {}), ...(query.originSiteId ? { originSiteId: query.originSiteId } : {}), pinned: page.pinned.map(decorate), sessions: page.sessions.map(decorate), }); } // Same-major compatibility: listSessions() has historically returned an // array. Preserve that wire shape while adding personal pin metadata/order; // cursor consumers opt into the additive page view. A query flag rather // than a /sessions/page path is deliberate: an older API safely ignores it // and returns its historical array instead of treating "page" as a UUID. return c.json([...page.pinned, ...page.sessions].map(decorate)); }); app.get("/v1/workspaces/:workspaceId/agent-topology", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); let authorizationScope; try { authorizationScope = await requireSessionAuthorizationListScope(deps, grant, "http"); } catch (error) { throw sessionAuthorizationHttpError(error); } const query = agentTopologyQuery(c.req.query()); const startedAtMs = performance.now(); const mode = query.subject ? "subject" : query.query ? "query" : "browse"; const metricAuthorizationScope = authorizationScope?.kind === "scoped" ? "scoped" : "workspace"; if ((query.query || query.subject) && !settings.workDiscoveryEnabled) { observeWorkDiscovery(deps.observability, { surface: "agent_topology", mode, outcome: "disabled", authorizationScope: metricAuthorizationScope, durationMs: performance.now() - startedAtMs, responseBytes: 0, resultCount: 0, overlapCount: 0, matchCounts: {}, }); throw new HTTPException(503, { message: "Agent work discovery is disabled by the operator.", }); } try { const orderBy: SessionDiscoveryOrderBy = query.query || query.subject ? "relevance" : "updatedAt"; const page = await listSessionDiscoverySummaries(db, workspaceId, { limit: query.limit, orderBy, subjectId: grant.subjectId, ...(query.cursor ? { cursor: query.cursor } : {}), ...(query.parentSessionId !== undefined ? { parentSessionId: query.parentSessionId } : {}), ...(query.rootSessionId ? { rootSessionId: query.rootSessionId } : {}), ...(query.query ? { query: query.query } : {}), ...(query.statuses ? { statuses: query.statuses } : {}), activeOnly: query.activeOnly, ...(query.recentHours !== undefined ? { recentHours: query.recentHours } : {}), ...(query.subject ? { subject: query.subject } : {}), ...(query.claimLimit !== undefined ? { claimLimit: query.claimLimit } : {}), includeWorkDiscovery: settings.workDiscoveryEnabled, ...(authorizationScope ? { authorizationScope } : {}), }); const ancestorPaths = query.query || query.subject ? await listSessionDiscoveryAncestorPaths( db, workspaceId, page.sessions.map((session) => session.id), authorizationScope ?? undefined, grant.subjectId, ) : new Map(); const sessions: AgentTopologyPageResponse["sessions"] = page.sessions.map((session) => { const blocker = session.effectiveControl.primaryBlocker; return { id: session.id, title: session.title, titleTruncated: session.titleOriginalChars !== null && session.titleOriginalChars > Array.from(session.title ?? "").length, parentSessionId: session.parentSessionId, rootSessionId: session.rootSessionId, nestedAgentDepth: session.nestedAgentDepth, ancestorPath: (ancestorPaths.get(session.id) ?? []).map((ancestor) => ({ id: ancestor.id, title: ancestor.title, titleTruncated: ancestor.titleOriginalChars !== null && ancestor.titleOriginalChars > Array.from(ancestor.title ?? "").length, })), status: session.status, goal: session.goal ? { status: session.goal.status, summary: session.goal.text, summaryTruncated: session.goal.textOriginalChars > Array.from(session.goal.text).length, } : null, pause: { state: session.effectiveControl.state, additionalBlockerCount: session.effectiveControl.additionalBlockerCount, source: blocker ? { kind: blocker.kind, ...(blocker.sessionId ? { sessionId: blocker.sessionId } : {}), displayName: blocker.displayName, displayNameTruncated: blocker.displayNameOriginalChars > Array.from(blocker.displayName).length, } : null, }, children: session.treeStats, relatedWork: session.workDiscovery, createdAt: session.createdAt, updatedAt: session.updatedAt, }; }); const response = { sessions, total: page.total, hasMore: page.hasMore, humanAdvisoriesEnabled: settings.workDiscoveryEnabled && settings.workDiscoveryHumanAdvisoriesEnabled, nextCursor: page.nextCursor ? encodeAgentTopologyCursor({ cursor: page.nextCursor, parentSessionId: query.parentSessionId === undefined ? "all" : query.parentSessionId, rootSessionId: query.rootSessionId ?? null, query: query.query ?? null, statuses: query.statuses ?? [], activeOnly: query.activeOnly, recentHours: query.recentHours ?? null, subject: query.subject ?? null, claimLimit: query.claimLimit ?? null, }) : null, } satisfies AgentTopologyPageResponse; observeWorkDiscovery(deps.observability, { surface: "agent_topology", mode, outcome: sessions.length === 0 ? "empty" : "ok", authorizationScope: metricAuthorizationScope, durationMs: performance.now() - startedAtMs, responseBytes: Buffer.byteLength(JSON.stringify(response), "utf8"), ...summarizeWorkDiscoveryRows(sessions), }); return c.json(response); } catch (error) { observeWorkDiscovery(deps.observability, { surface: "agent_topology", mode, outcome: "error", authorizationScope: metricAuthorizationScope, durationMs: performance.now() - startedAtMs, responseBytes: 0, resultCount: 0, overlapCount: 0, matchCounts: {}, }); throw error; } }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const session = await getSessionForSubject( db, workspaceId, sessionId, grant.subjectId, relatedSessionAccessFor(c), ); if (!session) { throw new HTTPException(404, { message: "session not found" }); } const activity = await backgroundCommandActivityForSessions(db, { accountId: grant.accountId, workspaceId, sessionIds: [sessionId], }); const scheduleTargets = hasPermission(grant.permissions, "scheduled_tasks:run") && hasPermission(grant.permissions, "sessions:control") ? await scheduledSessionIds(db, workspaceId, [sessionId]) : new Set(); return c.json( await withEffectivePolicy(deps, workspaceId, grant.subjectId, { ...session, hasSchedules: scheduleTargets.has(sessionId), ...(activity.get(sessionId) ? { backgroundCommandActivity: activity.get(sessionId) } : {}), }), ); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/model-context", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const session = await getSessionForSubject( db, workspaceId, sessionId, grant.subjectId, relatedSessionAccessFor(c), ); if (!session) { throw new HTTPException(404, { message: "session not found" }); } return c.json( await getLatestSessionModelContext(db, { accountId: grant.accountId, workspaceId, sessionId, }), ); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/background-commands", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const commands = await listSessionBackgroundCommands(db, { accountId: grant.accountId, workspaceId, sessionId, activeOnly: true, }); return c.json({ commands }); }); app.delete( "/v1/workspaces/:workspaceId/sessions/:sessionId/background-commands/:commandId", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const commandId = c.req.param("commandId"); if (!z.string().uuid().safeParse(commandId).success) { throw new HTTPException(404, { message: "background command not found" }); } const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const result = await requestSessionBackgroundCommandCancellation(db, { accountId: grant.accountId, workspaceId, sessionId, commandId, subjectId: grant.subjectId, }); if (!result.command) { throw new HTTPException(404, { message: "background command not found" }); } return c.json(result); }, ); app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/visibility", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const authorization = await requireAccessGrantAuthorization(c, deps, workspaceId); const parsed = UpdateSessionVisibilityRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new ApiHttpError(422, { code: "validation_failed", message: "Invalid session visibility request.", retryable: false, details: { fields: zodErrorFields(parsed.error) }, }); } try { return c.json( await updateManagedHumanSessionVisibility( deps, authorization, workspaceId, sessionId, parsed.data, "http", ), ); } catch (error) { throw sessionTenancyHttpError(error); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/forks", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const authorization = await requireAccessGrantAuthorization(c, deps, workspaceId); const parsed = ForkSessionRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new ApiHttpError(422, { code: "validation_failed", message: "Invalid session fork request.", retryable: false, details: { fields: zodErrorFields(parsed.error) }, }); } try { const response = await forkManagedHumanSession( deps, authorization, workspaceId, sessionId, parsed.data, "http", ); return c.json(response, response.replay ? 200 : 201); } catch (error) { throw sessionTenancyHttpError(error); } }); const publishRealtimeMutation = async ( accountId: string, workspaceId: string, sessionId: string, result: { eventIds: string[]; workflowWakeRevision: number | null; }, ): Promise => { const events = ( await Promise.all(result.eventIds.map((eventId) => getSessionEvent(db, workspaceId, eventId))) ).filter((event) => event !== null); await publishDurableSessionEvents(bus, workspaceId, sessionId, events); if (result.workflowWakeRevision !== null) { await workflowClient.wakeSessionWorkflow({ accountId, workspaceId, sessionId, workflowId: workflowIdForSession(sessionId), wakeRevision: result.workflowWakeRevision, }); } }; app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/realtime", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(400, { message: "invalid session id" }); } const parsed = BeginSessionRealtimeRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(400, { message: "invalid session realtime request", }); } try { const result = await withWorkspaceSessionActivityRls(db, workspaceId, async (scopedDb) => beginSessionRealtimeInTransaction(scopedDb, { accountId: grant.accountId, workspaceId, sessionId, ownerSubjectId: grant.subjectId, ...parsed.data, }), ); await publishRealtimeMutation(grant.accountId, workspaceId, sessionId, result); c.header("cache-control", "private, no-store"); return c.json({ mode: result.mode, replay: result.replay }, result.replay ? 200 : 201); } catch (error) { throw sessionRealtimeHttpError(error); } }); app.patch( "/v1/workspaces/:workspaceId/sessions/:sessionId/realtime/:realtimeId/heartbeat", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const realtimeId = c.req.param("realtimeId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if ( !z.string().uuid().safeParse(sessionId).success || !z.string().uuid().safeParse(realtimeId).success ) { throw new HTTPException(400, { message: "invalid realtime lifecycle id", }); } const parsed = RenewSessionRealtimeRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(400, { message: "invalid realtime heartbeat request", }); } try { const result = await withWorkspaceSessionActivityRls(db, workspaceId, async (scopedDb) => renewSessionRealtimeInTransaction(scopedDb, { workspaceId, sessionId, realtimeId, ownerSubjectId: grant.subjectId, ...parsed.data, }), ); await publishRealtimeMutation(grant.accountId, workspaceId, sessionId, result); c.header("cache-control", "private, no-store"); return c.json({ mode: result.mode, replay: result.replay }); } catch (error) { throw sessionRealtimeHttpError(error); } }, ); app.delete("/v1/workspaces/:workspaceId/sessions/:sessionId/realtime/:realtimeId", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const realtimeId = c.req.param("realtimeId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if ( !z.string().uuid().safeParse(sessionId).success || !z.string().uuid().safeParse(realtimeId).success ) { throw new HTTPException(400, { message: "invalid realtime lifecycle id", }); } const parsed = EndSessionRealtimeRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(400, { message: "invalid realtime end request", }); } try { const result = await withWorkspaceSessionActivityRls(db, workspaceId, async (scopedDb) => endSessionRealtimeInTransaction(scopedDb, { workspaceId, sessionId, realtimeId, ownerSubjectId: grant.subjectId, ...parsed.data, }), ); await publishRealtimeMutation(grant.accountId, workspaceId, sessionId, result); c.header("cache-control", "private, no-store"); return c.json({ mode: result.mode, replay: result.replay }); } catch (error) { throw sessionRealtimeHttpError(error); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/realtime/webrtc", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const parsed = CodexRealtimeWebrtcRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(422, { message: "invalid Codex realtime WebRTC request", }); } c.header("cache-control", "private, no-store"); try { const { realtimeId, operationId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, rotate, browserActivation, ...providerRequest } = parsed.data; const claim = await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => claimSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, operationId, ownerSubjectId: grant.subjectId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, rotate, promotionMode: browserActivation === "required" ? "staged" : "legacy", }), ), ); if (claim.replay) { if ( (claim.connection.state !== "ready" && claim.connection.state !== "active") || !claim.connection.sdpAnswer ) { throw new SessionRealtimeConflictError( "REALTIME_CONNECTION_STATE_CHANGED", "Realtime connection operation cannot be replayed; rotate with a new operation", ); } const legacyActivation = browserActivation !== "required" && claim.connection.state === "ready" ? await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => activateSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: claim.connection.id, operationId, ownerSubjectId: grant.subjectId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, connectionEpoch: claim.connection.connectionEpoch, }), ), ) : null; return c.json({ sdp: claim.connection.sdpAnswer, version: "v3" as const, model: "gpt-live-1-boulder-alpha" as const, connectionId: claim.connection.id, connectionEpoch: claim.connection.connectionEpoch, startupFenceSequence: claim.connection.startupFenceSequence, modeVersion: legacyActivation?.mode.version ?? claim.modeVersion, replay: true, }); } const broker = buildSessionCodexRealtimeBroker( db, settings, workspaceId, sessionId, deps.codexFetch, ); try { const answer = await broker({ request: providerRequest, signal: c.req.raw.signal, }); const completed = await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => completeSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: claim.connection.id, operationId, connectionEpoch: claim.connection.connectionEpoch, sdpAnswer: answer.sdp, }), ), ); const legacyActivation = browserActivation !== "required" ? await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => activateSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: completed.connection.id, operationId, ownerSubjectId: grant.subjectId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, connectionEpoch: completed.connection.connectionEpoch, }), ), ) : null; return c.json({ ...answer, connectionId: completed.connection.id, connectionEpoch: completed.connection.connectionEpoch, startupFenceSequence: completed.connection.startupFenceSequence, modeVersion: legacyActivation?.mode.version ?? claim.modeVersion, replay: false, }); } catch (error) { if (error instanceof CodexRealtimeBrokerError) { await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => failSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: claim.connection.id, operationId, connectionEpoch: claim.connection.connectionEpoch, failureCode: error.reason, }), ), ).catch(() => undefined); } throw error; } } catch (error) { if (error instanceof SessionRealtimeConflictError) { throw sessionRealtimeHttpError(error); } if (!(error instanceof CodexRealtimeBrokerError)) throw error; const failure = codexRealtimeHttpFailure(error); return c.json( { error: { status: failure.status, code: failure.code, message: error.message, retryable: failure.retryable, details: { reason: error.reason, ...(error.providerStatus === null ? {} : { providerStatus: error.providerStatus }), }, }, }, failure.status, ); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/realtime/gateway", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const parsed = GatewayRealtimeConnectRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(422, { message: "invalid Gateway realtime request", }); } c.header("cache-control", "private, no-store"); const { realtimeId, operationId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, rotate, } = parsed.data; let claim: Awaited> | null = null; let connectionCompleted = false; try { claim = await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => claimSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, operationId, ownerSubjectId: grant.subjectId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, rotate, promotionMode: "staged", }), ), ); if (claim.replay) { throw new SessionRealtimeConflictError( "REALTIME_CONNECTION_STATE_CHANGED", "Realtime Gateway tokens are single-use; reconnect with a new operation", ); } const secret = await createGatewayRealtimeConnectionSecret({ db, settings, workspaceId, sessionId, model: claim.mode.model, fetchImpl: deps.codexFetch ?? fetch, }); const claimed = claim; const completed = await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => completeSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: claimed.connection.id, operationId, connectionEpoch: claimed.connection.connectionEpoch, sdpAnswer: "gateway-client-secret-minted", }), ), ); connectionCompleted = true; return c.json({ ...secret, connectionId: completed.connection.id, connectionEpoch: completed.connection.connectionEpoch, startupFenceSequence: completed.connection.startupFenceSequence, modeVersion: claimed.modeVersion, replay: false as const, }); } catch (error) { if (claim !== null && !claim.replay && !connectionCompleted) { const claimed = claim; await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => failSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: claimed.connection.id, operationId, connectionEpoch: claimed.connection.connectionEpoch, failureCode: error instanceof GatewayRealtimeBrokerError ? error.code : "gateway_error", }), ), ).catch(() => undefined); } if (error instanceof SessionRealtimeConflictError) throw sessionRealtimeHttpError(error); if (!(error instanceof GatewayRealtimeBrokerError)) throw error; const status = error.code === "credential_unavailable" ? 409 : 502; return c.json( { error: { status, code: `GATEWAY_REALTIME_${error.code.toUpperCase()}`, message: error.message, retryable: error.code === "provider_error", }, }, status, ); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/realtime/supergrok", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const parsed = GatewayRealtimeConnectRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(422, { message: "invalid SuperGrok realtime request" }); } c.header("cache-control", "private, no-store"); const { realtimeId, operationId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, rotate, } = parsed.data; let claim: Awaited> | null = null; let connectionCompleted = false; try { claim = await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => claimSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, operationId, ownerSubjectId: grant.subjectId, browserInstanceId, ownerKey, expectedVersion, expectedConnectionEpoch, rotate, promotionMode: "staged", }), ), ); if (claim.replay) { throw new SessionRealtimeConflictError( "REALTIME_CONNECTION_STATE_CHANGED", "SuperGrok realtime tokens are single-use; reconnect with a new operation", ); } const secret = await createXaiRealtimeConnectionSecret({ db, settings, accountId: grant.accountId, workspaceId, subjectId: grant.subjectId, sessionId, model: claim.mode.model, fetchImpl: deps.xaiFetch ?? fetch, }); const claimed = claim; const completed = await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => completeSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: claimed.connection.id, operationId, connectionEpoch: claimed.connection.connectionEpoch, sdpAnswer: "supergrok-client-secret-minted", }), ), ); connectionCompleted = true; return c.json({ ...secret, connectionId: completed.connection.id, connectionEpoch: completed.connection.connectionEpoch, startupFenceSequence: completed.connection.startupFenceSequence, modeVersion: claimed.modeVersion, replay: false as const, }); } catch (error) { if (claim !== null && !claim.replay && !connectionCompleted) { const claimed = claim; await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => failSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId: claimed.connection.id, operationId, connectionEpoch: claimed.connection.connectionEpoch, failureCode: error instanceof XaiRealtimeBrokerError ? error.code : "supergrok_error", }), ), ).catch(() => undefined); } if (error instanceof SessionRealtimeConflictError) throw sessionRealtimeHttpError(error); if (!(error instanceof XaiRealtimeBrokerError)) throw error; const status = error.code === "credential_unavailable" ? 409 : 502; return c.json( { error: { status, code: `SUPERGROK_REALTIME_${error.code.toUpperCase()}`, message: error.message, retryable: error.code === "provider_error", }, }, status, ); } }); app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/realtime/:realtimeId/connections/:connectionId/activate", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const realtimeId = c.req.param("realtimeId"); const connectionId = c.req.param("connectionId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if ( !z.string().uuid().safeParse(sessionId).success || !z.string().uuid().safeParse(realtimeId).success || !z.string().uuid().safeParse(connectionId).success ) { throw new HTTPException(400, { message: "invalid realtime connection id", }); } const parsed = ActivateCodexRealtimeConnectionRequest.safeParse( await c.req.json().catch(() => null), ); if (!parsed.success) { throw new HTTPException(422, { message: "invalid realtime connection activation", }); } try { const result = await withWorkspaceRls(db, workspaceId, async (scopedDb) => scopedDb.transaction(async (tx) => activateSessionRealtimeConnectionInTransaction(tx as unknown as Database, { workspaceId, sessionId, realtimeId, connectionId, ownerSubjectId: grant.subjectId, ...parsed.data, }), ), ); c.header("cache-control", "private, no-store"); return c.json({ mode: result.mode, replay: result.replay }); } catch (error) { throw sessionRealtimeHttpError(error); } }, ); app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/realtime/:realtimeId/sync", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const realtimeId = c.req.param("realtimeId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:control", ); const grant = authorization.grant; const beforeCommit = externalContinuationCommitAuthorizer(authorization); const captureLinked = prepareExternalLinkTurnAdmission(authorization); if ( !z.string().uuid().safeParse(sessionId).success || !z.string().uuid().safeParse(realtimeId).success ) { throw new HTTPException(400, { message: "invalid realtime ledger id" }); } const parsed = SyncSessionRealtimeLedgerRequest.safeParse( await c.req.json().catch(() => null), ); if (!parsed.success) { throw new HTTPException(422, { message: "invalid realtime ledger sync request", }); } try { const result = await withWorkspaceSessionActivityRls(db, workspaceId, async (scopedDb) => { // Acquire origin authority before inference/session locks. A deferred // realtime session has no initial worker turn; capture its exact // linked actor when the ledger actually admits ordinary agent work. await beforeCommit?.(scopedDb as unknown as Database); return syncSessionRealtimeLedgerInTransaction( scopedDb, { workspaceId, sessionId, realtimeId, ownerSubjectId: grant.subjectId, ...parsed.data, controlLockTimeoutMs: workspaceControlRequestLockTimeoutMs(), }, captureLinked ? { afterDelegationAdmission: async ({ turnId }) => captureLinked(scopedDb as unknown as Database, sessionId, turnId), } : {}, ); }); await publishRealtimeMutation(grant.accountId, workspaceId, sessionId, result); c.header("cache-control", "private, no-store"); return c.json({ accepted: result.accepted, outbound: result.outbound }); } catch (error) { throw sessionRealtimeHttpError(error); } }, ); // Personal pin only: this is organization state for the authenticated member, // not a mutation of the shared session. It deliberately requires read access // (not session control) and returns 404 for a foreign/inaccessible session. app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/pin", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:read", ); const grant = authorization.grant; const sessionId = c.req.param("sessionId"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const parsed = UpdateSessionPinRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(400, { message: "invalid session pin request" }); } try { const session = await setSessionPin(db, { workspaceId, subjectId: grant.subjectId, sessionId, // Same owner-only personal-workspace fallback as the list above. personalWorkspaceOwnerException: hasVerifiedOwningUserAuthorization(authorization), ...parsed.data, }); if (!session) { throw new HTTPException(404, { message: "session not found" }); } return c.json( await withEffectivePolicy( deps, workspaceId, grant.subjectId, projectSessionForRelatedAccess(session, relatedSessionAccessFor(c)), ), ); } catch (error) { if (error instanceof SessionPinAccessError) { throw new HTTPException(403, { message: error.message }); } if (error instanceof SessionPinVersionConflictError) { return c.json( { message: "session pin changed in another client", current: error.current, }, 409, ); } throw error; } }); // Personal follow-up state. Session reads stay side-effect free; clients use // this explicit mutation after a deliberate read/unread action. The managed // web console treats an exact foreground chat event frontier as that action, // while the actively-working label remains independent of acknowledgment. app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/attention", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:read", ); const grant = authorization.grant; const sessionId = c.req.param("sessionId"); if (!z.string().uuid().safeParse(sessionId).success) { throw new HTTPException(404, { message: "session not found" }); } const parsed = UpdateSessionAttentionRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(400, { message: "invalid session attention request" }); } try { const session = await setSessionAttention(db, { workspaceId, subjectId: grant.subjectId, sessionId, personalWorkspaceOwnerException: hasVerifiedOwningUserAuthorization(authorization), ...parsed.data, }); if (!session) throw new HTTPException(404, { message: "session not found" }); return c.json( await withEffectivePolicy( deps, workspaceId, grant.subjectId, projectSessionForRelatedAccess(session, relatedSessionAccessFor(c)), ), ); } catch (error) { if (error instanceof SessionPinAccessError) { throw new HTTPException(403, { message: error.message }); } if (error instanceof SessionAttentionVersionConflictError) { return c.json( { message: "session attention changed in another client", current: error.current, }, 409, ); } throw error; } }); // Personal chat organization: archived roots leave the ordinary rail only // for this member and remain recoverable through the archived list view. app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/archive", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:read", ); const grant = authorization.grant; const sessionId = c.req.param("sessionId"); const parsed = UpdateSessionArchiveRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(400, { message: "invalid session archive request" }); } try { const session = await setSessionArchive(db, { workspaceId, subjectId: grant.subjectId, sessionId, personalWorkspaceOwnerException: hasVerifiedOwningUserAuthorization(authorization), ...parsed.data, }); if (!session) throw new HTTPException(404, { message: "session not found" }); return c.json( await withEffectivePolicy( deps, workspaceId, grant.subjectId, projectSessionForRelatedAccess(session, relatedSessionAccessFor(c)), ), ); } catch (error) { if (error instanceof SessionPinAccessError) { throw new HTTPException(403, { message: error.message }); } if (error instanceof SessionArchiveVersionConflictError) { return c.json( { message: "session archive changed in another client", current: error.current }, 409, ); } throw error; } }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/lineage", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const lineage = await readSessionLineage(deps, grant, c.req.param("sessionId")); const sessionIds = [ c.req.param("sessionId"), ...lineage.ancestors.map((session) => session.id), ]; const collect = (nodes: LineageNode[]) => { for (const node of nodes) { sessionIds.push(node.session.id); collect(node.children); } }; collect(lineage.children); const targets = hasPermission(grant.permissions, "scheduled_tasks:run") && hasPermission(grant.permissions, "sessions:control") ? await scheduledSessionIds(db, workspaceId, sessionIds) : new Set(); const decorateNodes = (nodes: LineageNode[]): LineageNode[] => nodes.map((node) => ({ ...node, session: { ...node.session, hasSchedules: targets.has(node.session.id) }, children: decorateNodes(node.children), })); const policy = await loadEffectivePolicyContext(deps, workspaceId, grant.subjectId); return c.json({ ...lineage, sessionHasSchedules: targets.has(c.req.param("sessionId")), ancestors: lineage.ancestors.map((session) => sessionWithEffectiveToolPolicy( { ...session, hasSchedules: targets.has(session.id) }, policy.workspaceServerIds, policy.workspaceDefaultServerIds, ), ), children: decorateNodes(mapLineageNodes(lineage.children, policy)), }); }); // Pin (or unpin) the session's Codex account. body { target: "auto" | "" }: // "auto" clears the pin (the session follows the workspace active pointer); a // uuid pins the session to that specific account. Overrides a capacity-blocked // turn; a running attempt keeps its account. 404 when the session or the target // account id isn't in the workspace. app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/codex-account", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); const body = (await c.req.json()) as { target?: string }; const target = typeof body.target === "string" ? body.target : ""; if (!target) { throw new HTTPException(400, { message: 'target is required ("auto" or an account id)', }); } if (!deps.sessionAuthorization) { try { await requireSessionAuthorization(deps, grant, { sessionId, operation: "session.codex_account.write", surface: "http", }); } catch (error) { throw sessionAuthorizationHttpError(error); } } const pinned = target === "auto" ? null : target; const mutation = await switchSessionCodexAccount(db, { workspaceId, sessionId, credentialId: pinned, subjectId: grant.subjectId, }); const ok = mutation.result.changed; if (!ok) { throw new HTTPException(404, { message: "session or codex account not found", }); } await Promise.allSettled( mutation.wakeTargets.map((wake) => workflowClient.signalCodexCapacity ? workflowClient.signalCodexCapacity({ accountId: wake.accountId, workspaceId: wake.workspaceId, sessionId: wake.sessionId, workflowId: wake.workflowId, wakeRevision: wake.wakeRevision, workflowWakeRevision: wake.workflowWakeRevision, }) : workflowClient.wakeSessionWorkflow({ accountId: wake.accountId, workspaceId: wake.workspaceId, sessionId: wake.sessionId, workflowId: wake.workflowId, wakeRevision: wake.workflowWakeRevision, }), ), ); await publishDurableSessionEvents(bus, workspaceId, sessionId, mutation.result.events); return c.json({ pinned: target === "auto" ? "auto" : target, appliedTo: mutation.result.appliedTo, }); }); // Re-file the session into a workspace channel (rail organization only; // null = back to the unfiled inbox). Shared, workspace-visible state, so it // requires sessions:control like rename. Returns the refreshed session. app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/channel", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); await requireSessionAuthorization(deps, grant, { sessionId, operation: "session.control", surface: "http", }); const payload = UpdateSessionChannelRequest.parse(await c.req.json()); try { const updated = await setSessionChannel(db, { workspaceId, sessionId, channelId: payload.channelId, }); if (!updated) { throw new HTTPException(404, { message: "session not found" }); } } catch (error) { if (error instanceof ChannelNotFoundError) { throw new HTTPException(422, { message: error.message }); } throw error; } const session = await getSessionForSubject( db, workspaceId, sessionId, grant.subjectId, relatedSessionAccessFor(c), ); if (!session) { throw new HTTPException(404, { message: "session not found" }); } return c.json(await withEffectivePolicy(deps, workspaceId, grant.subjectId, session)); }); // Replace the complete ordered Variable Set selection. The DB mutation // serializes with turn claim, rejects live/shared sandbox use, and requests a // cold rotation before the new environment can be materialized. app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/variable-sets", async (c) => { const workspaceId = c.req.param("workspaceId"); const sessionId = c.req.param("sessionId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); try { await requireSessionAuthorization(deps, grant, { sessionId, operation: "session.variable_sets.write", surface: "http", }); } catch (error) { throw sessionAuthorizationHttpError(error); } // Detach still requires attach authority: it changes which protected // resources the session may materialize. Non-empty selections additionally // require use authority through validateVariableSetAttachment. requirePermission(grant, "variable-sets:attach"); const parsed = UpdateSessionVariableSetsRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new ApiHttpError(422, { code: "validation_failed", message: "Invalid session Variable Set request.", retryable: false, details: { fields: zodErrorFields(parsed.error) }, }); } const variableSets: VariableSet[] = []; for (const variableSetId of parsed.data.variableSetIds) { variableSets.push( await validateVariableSetAttachment({ settings, db }, grant, workspaceId, variableSetId), ); } const result = await updateSessionVariableSets(db, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, variableSets: variableSets.map((variableSet) => ({ id: variableSet.id, name: variableSet.name, scope: variableSet.scope, })), }); if (result.status === "not_found") { throw new HTTPException(404, { message: "session not found" }); } if (result.status === "invalid_variable_sets") { throw new ApiHttpError(422, { code: "validation_failed", message: "One or more selected Variable Sets are no longer available.", retryable: false, outcomeUnknown: false, details: { variableSetIds: result.variableSetIds }, }); } if (result.status === "blocked") { const messages = { turn_in_flight: "Variable Sets can be changed only when the session has no accepted, queued, claimed, or pending work.", shared_sandbox_group: "Variable Sets cannot be changed while this session shares a sandbox; fork it into a separate session first.", live_sandbox_holders: "Close active terminal, desktop, and sandbox operations before changing Variable Sets.", } as const; throw new HTTPException(409, { message: messages[result.reason] }); } if (result.status === "updated") { await publishDurableSessionEvents(bus, workspaceId, sessionId, [result.event]); } const session = await getSessionForSubject( db, workspaceId, sessionId, grant.subjectId, relatedSessionAccessFor(c), ); if (!session) { throw new HTTPException(404, { message: "session not found" }); } return c.json(await withEffectivePolicy(deps, workspaceId, grant.subjectId, session)); }); // Manual rename. A user-set title is permanent: the db write is // unconditional (source='user'), so it always pins the session over later // agent writes. Returns the refreshed session, mirroring GET detail. app.patch("/v1/workspaces/:workspaceId/sessions/:sessionId", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = UpdateSessionRequest.parse(await c.req.json()); const titleUpdate = await updateSessionTitle(deps, grant, sessionId, payload.title, "user"); // A session-returning member route must preserve the caller's private pin // projection. Returning the generic mapSession() default here would reset a // pinned React consumer to false/version 0 after a harmless rename. const session = await getSessionForSubject( db, workspaceId, sessionId, grant.subjectId, titleUpdate.relatedSessionAccess, ); if (!session) { throw new HTTPException(404, { message: "session not found" }); } return c.json(await withEffectivePolicy(deps, workspaceId, grant.subjectId, session)); }); app.delete("/v1/workspaces/:workspaceId/sessions/:sessionId", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const deleted = await deleteSessionTreeIfQuiescent(db, { workspaceId, subjectId: grant.subjectId, sessionId: c.req.param("sessionId"), controlLockTimeoutMs: workspaceControlRequestLockTimeoutMs(), }); switch (deleted.status) { case "deleted": return c.json({ deletedSessionCount: deleted.deletedSessionCount }); case "not_found": throw new HTTPException(404, { message: "session not found" }); case "not_root": throw new HTTPException(409, { message: "delete the root session to remove the complete workstream", }); case "active_sessions": throw new HTTPException(409, { message: "cancel the workstream and wait for active turns to finish before deleting it", }); case "active_video_generations": throw new HTTPException(409, { message: "wait for active video generations to finish before deleting this workstream", }); case "active_background_commands": throw new HTTPException(409, { message: "pause or cancel this workstream's background commands before deleting it", }); case "live_sandboxes": throw new HTTPException(409, { message: "wait for the workstream's sandbox activity to finish draining before deleting it", }); case "externally_referenced": throw new HTTPException(409, { message: "this workstream has durable workspace outputs or independent forks; archive it instead", }); } }); app.patch( "/v1/workspaces/:workspaceId/sessions/:sessionId/mcp-servers/:serverId/approval-policy", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); const parsedServerId = SessionMcpServerId.safeParse(c.req.param("serverId")); const payload = UpdateSessionMcpApprovalPolicyRequest.safeParse( await c.req.json().catch(() => null), ); if (!parsedServerId.success || !payload.success) { throw new HTTPException(400, { message: "invalid MCP approval-policy request", }); } await assertSessionExists(db, workspaceId, sessionId); return c.json( await updateSessionMcpApprovalPolicy( deps, grant, sessionId, parsedServerId.data, payload.data.requireApproval, ), ); }, ); app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/tool-policy", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); const payload = UpdateSessionToolPolicyRequest.parse(await c.req.json().catch(() => null)); try { const session = await updateSessionToolPolicy(deps, grant, sessionId, payload); return c.json(await withEffectivePolicy(deps, workspaceId, grant.subjectId, session)); } catch (error) { if (error instanceof SessionToolPolicyVersionConflictError) { return c.json( { code: error.code, message: error.message, currentVersion: error.currentVersion, }, 409, ); } throw error; } }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/goal", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const goal = await getSessionGoalWithContinuation(db, workspaceId, sessionId); if (!goal) { throw new HTTPException(404, { message: "session goal not found" }); } return c.json(goal); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/goal/revisions", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); return c.json(await listSessionGoalRevisions(db, workspaceId, sessionId)); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/goal/revisions/page", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const parsedQuery = ListSessionGoalRevisionsQuery.safeParse({ limit: c.req.query("limit"), before: c.req.query("before"), }); if (!parsedQuery.success) { throw new HTTPException(400, { message: "invalid goal revision page query" }); } const query = parsedQuery.data; try { return c.json( await listSessionGoalRevisionPage(db, workspaceId, sessionId, { limit: query.limit, ...(query.before ? { before: query.before } : {}), }), ); } catch (error) { if (error instanceof SessionControlConflictError) { throw new HTTPException(409, { message: error.message, cause: error }); } throw error; } }); app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/goal/revisions/:revisionId/apply", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = ApplySessionGoalRevisionRequest.parse(await c.req.json()); const revision = await getSessionGoalRevision( db, workspaceId, sessionId, c.req.param("revisionId"), ); if (!revision || revision.disposition !== "proposed") { throw new HTTPException(404, { message: "goal rewrite proposal not found", }); } if (!goalProposalMatchesExpectedRevision(revision, payload.expectedObjectiveRevision)) { throw new HTTPException(409, { message: `goal proposal was based on objective revision ${revision.baseObjectiveRevision}; requested fence ${payload.expectedObjectiveRevision} cannot apply it`, }); } try { const { goal, workflowWakeRevision, events } = await upsertSessionGoalWithEvent(db, { accountId: grant.accountId, workspaceId, sessionId, text: revision.text, successCriteria: revision.successCriteria, rootConstraints: revision.rootConstraints, mutationPolicy: revision.mutationPolicy, expectedObjectiveRevision: payload.expectedObjectiveRevision, expectedGoalId: revision.goalId, changeKind: revision.changeKind, changeRationale: payload.rationale ?? `Applied goal proposal ${revision.id}`, sourceProposalId: revision.id, createdBy: "api", actor: "api", }); await publishDurableSessionEvents(bus, workspaceId, sessionId, events); if (workflowWakeRevision !== null) { await workflowClient.wakeSessionWorkflow({ accountId: grant.accountId, workspaceId, sessionId, workflowId: workflowIdForSession(sessionId), wakeRevision: workflowWakeRevision, }); } return c.json((await getSessionGoalWithContinuation(db, workspaceId, sessionId)) ?? goal); } catch (error) { if (error instanceof SessionControlConflictError) { throw new HTTPException(409, { message: error.message, cause: error, }); } throw error; } }, ); app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/goal/revisions/:revisionId/reject", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = RejectSessionGoalRevisionRequest.parse(await c.req.json()); try { const result = await rejectSessionGoalRevisionWithEvent(db, { accountId: grant.accountId, workspaceId, sessionId, revisionId: c.req.param("revisionId"), expectedObjectiveRevision: payload.expectedObjectiveRevision, rationale: payload.rationale, }); await publishDurableSessionEvents(bus, workspaceId, sessionId, result.events); return c.json({ revision: result.revision, replay: result.replay }); } catch (error) { if (error instanceof SessionControlConflictError) { throw new HTTPException(409, { message: error.message, cause: error }); } throw error; } }, ); app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/goal/revisions/:revisionId/rollback", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = RollbackSessionGoalRevisionRequest.parse(await c.req.json()); const revision = await getSessionGoalRevision( db, workspaceId, sessionId, c.req.param("revisionId"), ); if (!revision || revision.disposition !== "applied") { throw new HTTPException(404, { message: "applied goal revision not found" }); } try { const { goal, workflowWakeRevision, events } = await upsertSessionGoalWithEvent(db, { accountId: grant.accountId, workspaceId, sessionId, text: revision.text, successCriteria: revision.successCriteria, rootConstraints: revision.rootConstraints, mutationPolicy: revision.mutationPolicy, expectedObjectiveRevision: payload.expectedObjectiveRevision, expectedGoalId: revision.goalId, changeKind: "replacement", changeRationale: payload.rationale, rollbackOfRevisionId: revision.id, createdBy: "api", actor: "api", }); await publishDurableSessionEvents(bus, workspaceId, sessionId, events); if (workflowWakeRevision !== null) { await workflowClient.wakeSessionWorkflow({ accountId: grant.accountId, workspaceId, sessionId, workflowId: workflowIdForSession(sessionId), wakeRevision: workflowWakeRevision, }); } return c.json((await getSessionGoalWithContinuation(db, workspaceId, sessionId)) ?? goal); } catch (error) { if (error instanceof SessionControlConflictError) { throw new HTTPException(409, { message: error.message, cause: error }); } throw error; } }, ); app.patch("/v1/workspaces/:workspaceId/sessions/:sessionId/goal", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = UpdateSessionGoalRequest.parse(await c.req.json()); const existing = await getSessionGoal(db, workspaceId, sessionId); if (!existing) { throw new HTTPException(404, { message: "session goal not found" }); } if (!("status" in payload)) { try { const { goal, workflowWakeRevision, events } = await upsertSessionGoalWithEvent(db, { accountId: grant.accountId, workspaceId, sessionId, text: payload.text, successCriteria: payload.successCriteria !== undefined ? payload.successCriteria : existing.successCriteria, rootConstraints: payload.rootConstraints !== undefined ? payload.rootConstraints : existing.rootConstraints, maxAutoContinuations: existing.maxAutoContinuations, mutationPolicy: payload.mutationPolicy ?? existing.mutationPolicy, expectedObjectiveRevision: payload.expectedObjectiveRevision, changeKind: "replacement", changeRationale: payload.rationale, createdBy: "api", actor: "api", }); if (events.length > 0) { await bus.publish(workspaceId, sessionId, events); } if (workflowWakeRevision !== null) { await workflowClient.wakeSessionWorkflow({ accountId: grant.accountId, workspaceId, sessionId, workflowId: workflowIdForSession(sessionId), wakeRevision: workflowWakeRevision, }); } return c.json((await getSessionGoalWithContinuation(db, workspaceId, sessionId)) ?? goal); } catch (error) { if (error instanceof SessionControlConflictError) { throw new HTTPException(409, { message: error.message, cause: error, }); } throw error; } } if (existing.status === "completed") { throw new HTTPException(409, { message: "session goal is completed; set a new goal instead", }); } if (payload.status === "paused") { const { goal, events } = await setSessionGoalStatusWithEvent(db, workspaceId, sessionId, { status: "paused", ...(payload.rationale ? { rationale: payload.rationale } : {}), pausedReason: "api", event: { type: "goal.paused", actor: "api", reason: "api", ...(payload.rationale ? { rationale: payload.rationale } : {}), }, }); if (events.length > 0) { await bus.publish(workspaceId, sessionId, events); } return c.json((await getSessionGoalWithContinuation(db, workspaceId, sessionId)) ?? goal); } // Resume: only valid from paused; resets counters and re-arms the loop. if (existing.status !== "paused") { throw new HTTPException(409, { message: `session goal is ${existing.status}; only paused goals can be resumed`, }); } const { goal, changed, workflowWakeRevision, events } = await setSessionGoalStatusWithEvent( db, workspaceId, sessionId, { status: "active", event: { type: "goal.resumed", actor: "api" }, }, ); // `changed` guards the racing-PATCH case: both requests can pass the // status pre-check, but only the transition winner emits and wakes. if (changed) { if (events.length > 0) { await bus.publish(workspaceId, sessionId, events); } // signalWithStart restarts an eligible idle workflow so the durable goal // revision is evaluated. A closed workspace/session gate keeps the // revision inert until that gate's own Resume mutation commits its wake. if (workflowWakeRevision !== null) { await workflowClient.wakeSessionWorkflow({ accountId: grant.accountId, workspaceId, sessionId, workflowId: workflowIdForSession(sessionId), wakeRevision: workflowWakeRevision, }); } } return c.json((await getSessionGoalWithContinuation(db, workspaceId, sessionId)) ?? goal); }); app.delete("/v1/workspaces/:workspaceId/sessions/:sessionId/goal", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const { event } = await clearSessionGoal(db, workspaceId, sessionId); if (event) { try { await bus.publish(workspaceId, sessionId, [event]); } catch { console.warn("[api] cleared-goal live publish failed; durable event reconciles on replay", { errorClass: "EventPublishOperationError", errorCode: "cleared_goal_live_publish_failed", origin: "api", }); } } return c.body(null, 204); }); // Operator context controls (slash-command palette: /clear, /compact). These // are session/operator actions — NOT a structured channel to the agent. Both // require sessions:control. app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/context/clear", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); // Explicit confirm on the wire (literal true) — an empty/accidental POST // cannot wipe context. Mirrors the client-side confirm affordance. A // missing/false confirm is a client error (400), not a server fault. const clearBody = ClearSessionContextRequest.safeParse(await c.req.json().catch(() => ({}))); if (!clearBody.success) { throw new HTTPException(400, { message: "context clear requires an explicit { confirm: true }", }); } // The database checks this under workspace/session locks so a turn cannot // start between an API precheck and the history rewrite. const result = await clearSessionContext(db, { accountId: grant.accountId, workspaceId, sessionId, }).catch((error: unknown) => { if (error instanceof SessionContextBusyError) { throw new HTTPException(409, { message: error.message }); } throw error; }); await appendAndPublishEvents(db, bus, workspaceId, sessionId, [ { type: "session.context.cleared", payload: { clearedBy: "api", supersededItems: result.supersededItems, markerPosition: result.markerPosition, }, }, ]); return c.body(null, 204); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/context/compact", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); CompactSessionContextRequest.parse((await c.req.json().catch(() => ({}))) ?? {}); // /compact sets one durable request. The worker clears it only in the same // fenced transaction that installs replacement history, so failed or stale // attempts cannot lose the request. const requested = await requestSessionCompaction(db, workspaceId, sessionId); await workflowClient.wakeSessionWorkflow({ accountId: grant.accountId, workspaceId, sessionId, workflowId: requested.temporalWorkflowId, wakeRevision: requested.wakeRevision, }); return c.json({ status: "pending", message: "Compaction will run at the next safe boundary.", }); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/events", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const rawAfter = c.req.query("after"); const rawBefore = c.req.query("before"); const after = eventSequence(rawAfter, 0); const before = optionalEventSequence(rawBefore); const compact = compactEvents(c.req.query("compact")); const explicitReplay = rawAfter !== undefined || rawBefore !== undefined || compact; const mode = eventEnumValue( c.req.query("mode"), SessionEventReadMode, "mode", explicitReplay ? "forensic" : "monitoring", ); const latestRequested = eventEnumValue( c.req.query("latest"), SessionEventLatestClass, "latest", undefined, ); const latestClass = latestRequested === undefined ? undefined : sessionEventLatestClassToSemanticClass(latestRequested); const resultMode = eventEnumValue( c.req.query("resultMode") ?? c.req.query("result"), SessionEventResultMode, "resultMode", "events", ); if (resultMode === "compact" && latestClass === undefined) { throw new HTTPException(400, { message: "resultMode=compact requires latest", }); } if ( latestClass && ["includeTypes", "excludeTypes", "includeClasses", "excludeClasses"].some( (name) => c.req.query(name) !== undefined, ) ) { throw new HTTPException(400, { message: "latest cannot be combined with event filters", }); } const direction = latestClass ? "before" : eventEnumValue( c.req.query("direction"), SessionEventReadDirection, "direction", before !== undefined ? "before" : rawAfter !== undefined ? "after" : mode === "monitoring" ? "before" : "after", ); const payloadMode = eventEnumValue( c.req.query("payloadMode"), SessionEventPayloadMode, "payloadMode", mode === "monitoring" ? "summary" : "full", ); const includeTypes = eventEnumList( c.req.query("includeTypes"), SessionEventType, "includeTypes", ); const excludeTypes = eventEnumList( c.req.query("excludeTypes"), SessionEventType, "excludeTypes", ); const includeClasses = eventEnumList( c.req.query("includeClasses"), SessionEventSemanticClass, "includeClasses", ); const excludeClasses = eventEnumList( c.req.query("excludeClasses"), SessionEventSemanticClass, "excludeClasses", ); const limit = latestClass ? 1 : eventListLimit( c.req.query("limit"), compact ? 5000 : mode === "monitoring" ? 250 : 2000, mode === "monitoring" ? 40 : 500, ); const dbPayloadMode = resultMode === "compact" ? ("full" as const) : payloadMode; const dbPage = await listSessionEventPage(db, workspaceId, sessionId, { after, ...(before !== undefined ? { before } : {}), limit, direction, payloadMode: dbPayloadMode, includeTypes, excludeTypes, includeClasses: latestClass ? [latestClass] : includeClasses, excludeClasses, ...(mode === "monitoring" ? { defaultExcludeTypes: SESSION_EVENT_RAW_DELTA_TYPES } : {}), ...(latestClass ? { authoritativeLatest: true } : {}), }); const events = dbPage.events; if (resultMode === "compact") { const event = events[0]; c.header("X-OpenGeni-Event-Result-Mode", "compact"); c.header("X-OpenGeni-Event-Result", event ? "found" : "not_found"); c.header("X-OpenGeni-Event-Mode", mode); c.header("X-OpenGeni-Event-Direction", direction); c.header("X-OpenGeni-Payload-Mode", "full"); c.header("X-OpenGeni-Forensic-Exact", "false"); if (!event) return c.json(null, 200); const result = compactSessionEventResult( event, latestClass!, dbPage.coveredSequence ?? { first: event.sequence, last: event.sequence, }, ); c.header("X-OpenGeni-Covered-First", String(result.coveredSequence.first)); c.header("X-OpenGeni-Covered-Last", String(result.coveredSequence.last)); return c.json(result); } const compactProjection = compact ? coalesceSessionEventDeltasWithCoverage(events) : null; const projected = compactProjection?.events ?? events; const forensicExact = !compact && mode === "forensic" && payloadMode === "full" && dbPage.fullPayloadsExact; const page = boundSessionEventHttpPage(projected, { direction, eventProjection: payloadMode === "full" ? "exact" : "bounded", ...(compactProjection ? { coveredThroughBySequence: compactProjection.coveredThroughBySequence } : {}), }); const hasMore = dbPage.hasMore || page.truncated; c.header("X-OpenGeni-Page-Bytes", String(page.bytes)); // One oversized exact event is admitted alone; never advertise a maximum // smaller than the response we actually deliver. c.header("X-OpenGeni-Page-Max-Bytes", String(Math.max(1024 * 1024, page.bytes))); c.header("X-OpenGeni-Page-Truncated", String(hasMore)); c.header("X-OpenGeni-Has-More", String(hasMore)); c.header("X-OpenGeni-Event-Mode", mode); c.header("X-OpenGeni-Event-Direction", direction); c.header("X-OpenGeni-Payload-Mode", payloadMode); c.header("X-OpenGeni-Forensic-Exact", String(forensicExact)); const coveredFirst = page.events[0]?.sequence; const coveredLastEvent = page.events.at(-1); const coveredLast = coveredLastEvent === undefined ? undefined : (compactProjection?.coveredThroughBySequence.get(coveredLastEvent.sequence) ?? coveredLastEvent.sequence); if (coveredFirst !== undefined) c.header("X-OpenGeni-Covered-First", String(coveredFirst)); if (coveredLast !== undefined) c.header("X-OpenGeni-Covered-Last", String(coveredLast)); const truncatedBy = page.truncated ? "http_bytes" : dbPage.truncatedBy; if (truncatedBy) c.header("X-OpenGeni-Truncated-By", truncatedBy); if (page.nextSequence !== null) { c.header( direction === "before" ? "X-OpenGeni-Next-Before" : "X-OpenGeni-Next-After", String(page.nextSequence), ); } return c.json(page.events); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/events/stream", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); let authorization; try { authorization = await requireSessionAuthorization(deps, grant, { sessionId, operation: "session.stream.read", surface: "stream", }); } catch (error) { throw sessionAuthorizationHttpError(error); } await assertSessionExists(db, workspaceId, sessionId); const after = Number(c.req.query("after") ?? c.req.header("Last-Event-ID") ?? 0); return sseSessionStream( db, bus, workspaceId, sessionId, Number.isFinite(after) ? after : 0, c.req.raw.signal, { ...browserSseDeliveryOptions(c.req.query("transport")), observability: deps.observability, actorEpoch: getManagedAuthRequestActorEpoch(c.req.raw) ?? undefined, reauthorizeAfterMs: authorization?.reauthorizeAfterMs ?? SESSION_AUTHORIZATION_DEFAULT_REAUTHORIZE_MS, reauthorize: async () => { const freshGrant = await requireFreshAccessGrant(c, deps, workspaceId, "sessions:read"); await requireSessionAuthorization(deps, freshGrant, { sessionId, operation: "session.stream.read", surface: "stream", }); }, }, ); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/turns", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); // Exact turn that most recently emitted durable `turn.started` — the same // boundary goal continuations use for inherited model/effort. Queued-only // or preflight-rejected turns are deliberately excluded. if (c.req.query("latestStarted") === "1" || c.req.query("latestStarted") === "true") { const latest = await getLatestStartedSessionTurn(db, workspaceId, sessionId); return c.json(latest ? [latest] : []); } return c.json( await listSessionTurns(db, workspaceId, sessionId, boundedLimit(c.req.query("limit"))), ); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/mcp-credentials/rotate", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:control", ); const payload = await c.req.json().catch(() => null); c.header("cache-control", "private, no-store"); return c.json( await rotateSessionMcpCredentialsForRequest( deps, authorization, c.req.param("sessionId"), payload, (tx) => requireFreshAccessGrant(c, { ...deps, db: tx }, workspaceId, "sessions:control"), ), ); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/queue", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); const snapshot = await getSessionQueueSnapshot(db, workspaceId, sessionId); if (!snapshot) throw new HTTPException(404, { message: "session not found" }); return c.json(projectQueueSnapshot(snapshot, sessionId, relatedSessionAccessFor(c))); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/queue/:turnId/move", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = MoveSessionQueueItemRequest.parse(await c.req.json()); try { const response = await moveHumanQueuePrompt( deps, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, }, c.req.param("turnId"), payload, ); return c.json({ ...response, snapshot: projectQueueSnapshot(response.snapshot, sessionId, relatedSessionAccessFor(c)), }); } catch (error) { return commandConflictResponse(c, error); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/queue/:turnId/edit", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = EditSessionQueueItemRequest.parse(await c.req.json()); try { const response = await editHumanQueuePrompt( deps, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, }, c.req.param("turnId"), payload, ); return c.json({ ...response, snapshot: projectQueueSnapshot(response.snapshot, sessionId, relatedSessionAccessFor(c)), }); } catch (error) { return commandConflictResponse(c, error); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/queue/:turnId/steer", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = SteerSessionQueueItemRequest.parse(await c.req.json()); try { const response = await steerHumanQueuePrompt( deps, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, }, c.req.param("turnId"), payload, ); return c.json({ ...response, snapshot: projectQueueSnapshot(response.snapshot, sessionId, relatedSessionAccessFor(c)), }); } catch (error) { return commandConflictResponse(c, error); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/queue/:turnId/delete", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = DeleteSessionQueueItemRequest.parse(await c.req.json()); try { const response = await deleteHumanQueuePrompt( deps, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, }, c.req.param("turnId"), payload, ); return c.json({ ...response, snapshot: projectQueueSnapshot(response.snapshot, sessionId, relatedSessionAccessFor(c)), }); } catch (error) { return commandConflictResponse(c, error); } }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/composer-draft", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); return c.json( await getHumanComposerDraft(deps, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, }), ); }); app.put("/v1/workspaces/:workspaceId/sessions/:sessionId/composer-draft", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); const sessionId = c.req.param("sessionId"); const payload = SaveComposerDraftRequest.parse(await c.req.json()); try { return c.json( await saveHumanComposerDraft( deps, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, }, payload, ), ); } catch (error) { return commandConflictResponse(c, error); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/control", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:control"); if (workspaceControlUtf8Bytes(grant.subjectId) > WORKSPACE_CONTROL_ACTOR_MAX_BYTES) { throw new HTTPException(400, { message: "workspace-control actor is too large", }); } const sessionId = c.req.param("sessionId"); const parsed = SessionControlRequest.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) { throw new HTTPException(400, { message: "invalid session control request", }); } try { const response = await controlHumanSessionWorkstream( deps, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, }, parsed.data, ); return c.json({ ...response, effectiveControl: projectEffectiveControlForRelatedAccess( response.effectiveControl, sessionId, relatedSessionAccessFor(c), ), }); } catch (error) { return commandConflictResponse(c, error); } }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/steer", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:control", ); const grant = authorization.grant; const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = parseSteerSessionAdmission(await c.req.json().catch(() => null)); let result: Awaited>; try { result = await acceptSessionUserMessage(deps, grant, workspaceId, sessionId, { text: payload.text, annotations: payload.annotations, modelContext: payload.modelContext ?? null, resources: payload.resources, model: payload.model ?? null, reasoningEffort: payload.reasoningEffort ?? null, latencyMode: payload.latencyMode ?? null, mcpCredentialUpdates: payload.mcpCredentialUpdates ?? [], connectionAuthorities: payload.connectionAuthorities, ...(payload.selectedHostMcpDelegations ? { selectedHostMcpDelegations: payload.selectedHostMcpDelegations } : {}), ...(payload.personalResourceAttachment ? { personalResourceAttachment: payload.personalResourceAttachment } : {}), authorization, delivery: "steer", origin: "human", ...(payload.controlEtag !== undefined ? { controlEtag: payload.controlEtag } : {}), ...(payload.expectedDraftRevision !== undefined ? { expectedDraftRevision: payload.expectedDraftRevision } : {}), ...(payload.clientEventId ? { clientEventId: payload.clientEventId } : {}), }); } catch (error) { return commandConflictResponse(c, error); } return c.json(result, 202); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/composer-draft/submit", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:control", ); const grant = authorization.grant; const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const payload = SubmitComposerDraftRequest.parse(await c.req.json().catch(() => null)); let result: Awaited>; try { result = await submitComposerDraftForRequest(deps, grant, workspaceId, sessionId, payload, { authorization, }); } catch (error) { return commandConflictResponse(c, error); } return c.json(result, 202); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/events", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "sessions:control", ); const grant = authorization.grant; const sessionId = c.req.param("sessionId"); const event = parseSessionEventAdmission(await c.req.json().catch(() => null)); const refinedOperation = event.type === "user.approvalDecision" ? "session.approval.write" : event.type === "user.humanInputResponse" ? "session.human_input.write" : null; if (refinedOperation) { try { await requireSessionAuthorization(deps, grant, { sessionId, operation: refinedOperation, surface: "http", }); } catch (error) { throw sessionAuthorizationHttpError(error); } } if (event.type === "user.message") { let result: Awaited>; try { result = await acceptSessionUserMessage(deps, grant, workspaceId, sessionId, { text: event.payload.text, annotations: event.payload.annotations, modelContext: event.payload.modelContext ?? null, resources: event.payload.resources ?? [], model: event.payload.model ?? null, reasoningEffort: event.payload.reasoningEffort ?? null, latencyMode: event.payload.latencyMode ?? null, mcpCredentialUpdates: event.payload.mcpCredentialUpdates ?? [], connectionAuthorities: event.payload.connectionAuthorities, ...(event.payload.selectedHostMcpDelegations ? { selectedHostMcpDelegations: event.payload.selectedHostMcpDelegations } : {}), ...(event.payload.personalResourceAttachment ? { personalResourceAttachment: event.payload.personalResourceAttachment } : {}), authorization, ...(event.payload.controlEtag !== undefined ? { controlEtag: event.payload.controlEtag } : {}), ...(event.payload.expectedDraftRevision !== undefined ? { expectedDraftRevision: event.payload.expectedDraftRevision } : {}), ...(event.clientEventId ? { clientEventId: event.clientEventId } : {}), }); } catch (error) { return commandConflictResponse(c, error); } return c.json(result.accepted, 202); } if (event.type === "user.approvalDecision") { const accepted = await acceptSessionApprovalDecision(db, { accountId: grant.accountId, workspaceId, sessionId, subjectId: grant.subjectId, respondedByKind: childRequiresActionRespondedByKindForGrant(grant), payload: event.payload, clientEventId: event.clientEventId ?? null, }); if (accepted.action === "conflict") { throw new HTTPException(409, { message: `session is ${accepted.sessionStatus}; no unhandled approval is pending`, }); } await publishDurableSessionEvents(bus, workspaceId, sessionId, accepted.events); const workflowId = workflowIdForSession(sessionId); await workflowClient.signalApprovalDecision({ accountId: grant.accountId, workspaceId, sessionId, eventId: accepted.event.id, workflowId, workflowWakeRevision: accepted.workflowWakeRevision, }); return c.json(accepted.event, 202); } if (event.type === "user.humanInputResponse") { let accepted; try { accepted = await acceptSessionHumanInputResponse(db, { accountId: grant.accountId, workspaceId, sessionId, requestId: event.payload.requestId, response: event.payload.response, respondedBy: grant.subjectId, canonicalHumanSession: authorization.canonicalManagedHumanSession || authorization.canonicalLocalHumanSession, respondedByKind: childRequiresActionRespondedByKindForGrant(grant), clientEventId: event.clientEventId ?? null, }); } catch (error) { if (error instanceof SkillHumanResponseError) { throw new HTTPException( error.code === "conflict" ? 409 : error.code === "forbidden" ? 403 : 422, { message: error.message, }, ); } if (error instanceof HumanInputResponseValidationError) { throw new HTTPException( error.code === "HUMAN_AUTH_REQUIRED" ? 403 : error.code === "SKIP_NOT_ALLOWED" ? 409 : 422, { message: error.message, }, ); } throw error; } if (accepted.action === "not_found") { throw new HTTPException(404, { message: "human-input request not found", }); } await publishDurableSessionEvents(bus, workspaceId, sessionId, accepted.events); if (accepted.workflowWakeRevision !== null) { await workflowClient.signalApprovalDecision({ accountId: grant.accountId, workspaceId, sessionId, eventId: accepted.events[0]?.id ?? event.payload.requestId, workflowId: workflowIdForSession(sessionId), workflowWakeRevision: accepted.workflowWakeRevision, }); } if (accepted.action === "conflict") { throw new HTTPException(409, { message: "human-input request is not currently actionable", }); } return c.json(accepted.event, accepted.action === "completed" ? 200 : 202); } }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/human-input-requests", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); await assertSessionExists(db, workspaceId, sessionId); const rawStatus = c.req.query("status"); const status = rawStatus ? HumanInputRequestStatus.safeParse(rawStatus) : null; if (status && !status.success) { throw new HTTPException(400, { message: "invalid human-input request status", }); } const requests = await listSessionHumanInputRequests(db, workspaceId, sessionId, { ...(status?.success ? { status: status.data } : {}), }); return c.json({ requests }); }); app.get( "/v1/workspaces/:workspaceId/sessions/:sessionId/human-input-requests/:requestId", async (c) => { const workspaceId = c.req.param("workspaceId"); await requireAccessGrant(c, deps, workspaceId, "sessions:read"); const sessionId = c.req.param("sessionId"); const request = await getSessionHumanInputRequest( db, workspaceId, sessionId, c.req.param("requestId"), ); if (!request) throw new HTTPException(404, { message: "human-input request not found", }); return c.json(request); }, ); // ── API-direct stream capabilities + viewer attach (P1.4) ───────────────── // // All IN-PROCESS: capability negotiation reads the descriptor + the group // lease (liveness/epoch); viewer attach acquires a holder on the group lease // and (when cold) spins the box up via resume-by-id — NO worker, NO Temporal. // Gated behind sandboxOwnershipEnabled (the lease is inert with the flag off). // // ROUTE DISCIPLINE: requireAccessGrant BEFORE any Zod parse; explicit // HTTPException(400) on a parse failure (never a raw ZodError → 500); // HTTPException(409) on an epoch fence. function assertOwnershipEnabled(): void { if (!settings.sandboxOwnershipEnabled) { // The viewer-holder lifecycle rides the sandbox lease, which is dormant // until the flag flips per-environment. A 404 (not 403) keeps the route // invisible while disabled — it does not exist for this deployment yet. throw new HTTPException(404, { message: "sandbox ownership is not enabled for this deployment", }); } } // Resolve the shared-exposure disclosure for a session's group: `shared` when // the group has >1 session (addendum E.1), and the OTHER sessions' ids ONLY // (never their conversation/metadata; the query selects only id — stress g). async function resolveSharedExposure( workspaceId: string, session: { id: string; sandboxGroupId: string }, ): Promise<{ shared: boolean; sharedSessionIds: string[] }> { const ids = await listSessionIdsInGroup(db, workspaceId, session.sandboxGroupId); const others = ids.filter((id) => id !== session.id); return { shared: others.length > 0, sharedSessionIds: others }; } // GET .../stream-capabilities — the capability-negotiation read. Returns the // SessionCapabilities doc (descriptor + lease liveness/epoch + os + the // shared-exposure disclosure + the calling principal's acknowledgment state), // API-direct. It is a pure descriptor read: every URL/token stays null until an // exact, permission-checked POST /viewers grant mints it just in time. app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/stream-capabilities", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "sessions:read"); assertOwnershipEnabled(); const sessionId = c.req.param("sessionId"); const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } // Capability truth follows the ACTIVE placement, not the session's managed // home lease. A connected machine has no Modal group lease; treating that // absence as `cold` makes the client suppress the very viewer attach that // would mint its relay terminal/desktop cells. Resolve the active target once // and use the selfhosted liveness probe when it is a connected machine. const activeSandbox = session.activeSandboxId ? await getSandbox( db, { accountId: grant.accountId, workspaceId, subjectId: grant.subjectId }, session.activeSandboxId, ) : null; const selfhostedActive = activeSandbox?.kind === "selfhosted"; const lease = await readGroupLease( { db, settings }, { workspaceId, sandboxGroupId: session.sandboxGroupId }, ); const { shared, sharedSessionIds } = await resolveSharedExposure(workspaceId, session); const visibleSharedSessionIds = relatedSessionAccessFor(c) === "root" ? sharedSessionIds : []; // Per-principal acknowledgment: A acknowledging does not consent for B. The // Surface consent state so the UI can decide whether an explicit desktop // grant may be requested. This descriptor read never mints a credential; the // POST /viewers grant below re-checks both consent bits and stream:view. const ack = await getStreamAcknowledgment(db, { workspaceId, sandboxGroupId: session.sandboxGroupId, subjectId: grant.subjectId, }); const acknowledged = ack ? ack.acknowledgedUnredacted && (!shared || ack.acknowledgedShared) : false; // This GET is deliberately descriptor-only: no provider resume, display/ttyd // startup, port exposure, or short-lived bearer mint. Besides enforcing least // privilege, that keeps the 120-second stream credentials from aging before a // user opens their surface. POST /viewers is the sole credential grant. const commonNegotiation = { sessionId, desktopEnabled: settings.sandboxDesktopEnabled, // Human take-control: when the desktop is available + this policy is on // (default), the cell is mode "interactive" — the noVNC viewer drives :0 // (x11vnc runs without -viewonly). Off → mode "read-only" (client disables // take-control). Agent interaction uses managed ComputerSession tools. desktopInteractive: settings.sandboxDesktopInteractive, // Graceful degrade when scoped stream credentials cannot be minted. streamTokenSecretAvailable: !streamTokenDegraded(settings), desktopAcknowledged: acknowledged, shared, sharedSessionIds: visibleSharedSessionIds, terminalEnabled: settings.sandboxTerminalEnabled, } as const; let capabilities; if (selfhostedActive && activeSandbox.enrollmentId) { const [enrollment, liveConnection] = await Promise.all([ getEnrollment(db, grant, activeSandbox.enrollmentId), getLiveEnrollmentConnection(db, grant, activeSandbox.enrollmentId), ]); const reportedWorkspaceRoot = liveConnection?.workspaceRoot ?? enrollment?.workspaceRoot; const effectiveWorkspaceRoot = reportedWorkspaceRoot ? resolveConnectedMachineWorkspaceRoot(reportedWorkspaceRoot, session.workingDir) : null; let probeResponded = false; if (liveConnection?.connectionInstanceId) { const machine = new SelfhostedSession({ workspaceId: activeSandbox.workspaceId, agentId: liveConnection.id, connectionInstanceId: liveConnection.connectionInstanceId, // Capability negotiation only pings, so a pre-root agent may still // report upgrade guidance without exposing a false filesystem root. workspaceRoot: effectiveWorkspaceRoot ?? liveConnection.workspaceRoot ?? "/", controlRpc: new NatsControlRpc(async () => bus.getRequestConnection()), relay: relayConfigFromSettings(settings), epoch: session.activeEpoch, timeoutMs: settings.sandboxSelfhostedControlTimeoutMs, }); try { probeResponded = await machine.ping(); } catch { probeResponded = false; } } capabilities = await negotiateSelfhostedCapabilities({ ...commonNegotiation, os: enrollment?.os ?? session.sandboxOs, leaseEpoch: session.activeEpoch, enrollment: enrollment && effectiveWorkspaceRoot ? { ...enrollment, workspaceRoot: effectiveWorkspaceRoot } : enrollment, probeResponded, }); } else { capabilities = negotiateCapabilities({ ...commonNegotiation, backend: session.sandboxBackend as SandboxBackend, os: session.sandboxOs, liveness: lease?.liveness ?? "cold", leaseEpoch: lease?.leaseEpoch ?? 0, workspaceGeneration: lease?.workspaceGeneration ?? null, archiveGeneration: lease?.archiveGeneration ?? null, archiveComplete: lease?.archiveComplete ?? false, }); } const repositoryRoots = [ ...new Set( session.resources.flatMap((resource) => resource.kind === "repository" && typeof resource.mountPath === "string" ? [resource.mountPath.replace(/^\/+|\/+$/g, "")] : [], ), ), ].filter(Boolean); // SWAP-CASE desktop transport (BOTH directions): negotiateCapabilities keyed on // the HOME backend, but the pixel plane actually runs on the ACTIVE sandbox — and // the two backends use DIFFERENT wire transports. The advertised transport MUST // match where mintDesktopStream routed the pixels (relay IFF the active sandbox is // a selfhosted machine), or the client picks the wrong renderer and the socket // closes before it opens: // • modal-HOME swapped ONTO a selfhosted machine: negotiate says vnc-ws, but the // machine's desktop is the RELAY framebuffer (PNG-per-frame) → flip to // relay-frames/frames. (#171) // • selfhosted-HOME swapped AWAY to the cloud group box (activeSandboxId=null OR a // non-selfhosted active sandbox): negotiate says relay-frames (home=selfhosted), // but there is NO relay producer on the Modal box → the client hangs on a dead // relay socket ("desktop stream closed before it opened"). Flip to the Modal // noVNC/RFB tunnel (vnc-ws/novnc). This is the mirror of #171 and the missing // half that this fixes. // The single invariant: advertise relay-frames IFF (activeSandboxId set AND the // active sandbox kind is "selfhosted") — EXACTLY mintDesktopStream's routing. When // the desktop is available we set the transport from the ACTIVE sandbox in one // place (resolveActiveDesktopTransport), covering BOTH swap directions. let responseCapabilities = { ...capabilities, FileSystem: { ...capabilities.FileSystem, root: selfhostedActive ? capabilities.FileSystem.root : selectBackend( (activeSandbox?.kind === "modal" ? "modal" : session.sandboxBackend === "selfhosted" && settings.sandboxBackend !== "selfhosted" && settings.sandboxBackend !== "none" ? settings.sandboxBackend : session.sandboxBackend) as SandboxBackend, ).workspaceRoot, }, Git: { ...capabilities.Git, repos: capabilities.Git.available ? repositoryRoots : [], }, }; if (capabilities.DesktopStream.transport !== null) { const wire = resolveActiveDesktopTransport( selfhostedActive, settings.sandboxDesktopInteractive !== false, ); responseCapabilities = { ...responseCapabilities, DesktopStream: { ...capabilities.DesktopStream, ...wire }, }; } if (selfhostedActive) { const currentPointer = await readActiveSandbox(db, workspaceId, sessionId); if ( !currentPointer || currentPointer.activeSandboxId !== session.activeSandboxId || currentPointer.activeEpoch !== session.activeEpoch || currentPointer.workingDir !== session.workingDir ) { throw new ApiHttpError(409, { code: "conflict", message: "sandbox route changed while capabilities were being negotiated; retry", retryable: true, }); } } return c.json(responseCapabilities); }); // POST .../stream-capabilities/acknowledge — record the calling principal's // acknowledgment of the un-redacted pixel plane (and, when shared, the // shared-exposure disclosure). Reuses the acknowledgment machinery — gated on // stream:acknowledge, no new permission. Until this is recorded the // desktop-stream (viewer attach) path returns 409 (P3.2 consent gate). app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/stream-capabilities/acknowledge", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "stream:acknowledge"); assertOwnershipEnabled(); const sessionId = c.req.param("sessionId"); const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } const parsed = AcknowledgeStreamRequest.safeParse(await c.req.json().catch(() => ({}))); if (!parsed.success) { throw new HTTPException(400, { message: "invalid stream acknowledgment request", }); } const recorded = await recordStreamAcknowledgment(db, { accountId: grant.accountId, workspaceId, sandboxGroupId: session.sandboxGroupId, subjectId: grant.subjectId, acknowledgeUnredacted: parsed.data.acknowledgeUnredacted, acknowledgeShared: parsed.data.acknowledgeShared, }); return c.json({ acknowledged: recorded.acknowledgedUnredacted, acknowledgedShared: recorded.acknowledgedShared, }); }, ); // POST .../viewers — acquire a viewer holder on the desktop-stream (un-redacted // pixel) path. Gated on stream:view (strictly broader than sessions:read: the // pixel plane is un-redacted). THE CONSENT GATE: until the calling principal // has acknowledged the un-redacted plane this returns 409 // stream_acknowledgment_required; when the box is shared and the shared-exposure // disclosure is not acknowledged it returns 409 shared_acknowledgment_required. // Only after consent does it acquire the holder (spinning the box up in-process // when cold). app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/viewers", async (c) => { const workspaceId = c.req.param("workspaceId"); // Authenticate and bind the workspace before parsing. The requested plane // determines the narrower permission below: terminal-only holders must not // require the strictly broader un-redacted Desktop permission. const grant = await requireAccessGrant(c, deps, workspaceId); assertOwnershipEnabled(); const sessionId = c.req.param("sessionId"); const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } const parsed = AttachViewerRequest.safeParse(await c.req.json().catch(() => ({}))); if (!parsed.success) { throw new HTTPException(400, { message: "invalid viewer attach request", }); } // Resolve exact requested planes before authorization. Empty and // `desktop:false` v1 bodies remain terminal-only during rolling upgrades; // either new plane flag selects exact semantics. This closes desktop→terminal // privilege bleed and lets file edits avoid unrelated terminal bearers. const wantDesktop = parsed.data.desktop ?? false; const wantFiles = parsed.data.files ?? false; const hasExactPlaneSet = parsed.data.terminal !== undefined || parsed.data.files !== undefined; // v1 clients sent only `desktop:false` for both terminal and file warming; // retain its terminal-only grant during rolling upgrades. Presence of either // v2 flag switches to exact-plane semantics. const wantTerminal = parsed.data.terminal ?? (!hasExactPlaneSet && !wantDesktop); if (!wantDesktop && !wantTerminal && !wantFiles) { throw new HTTPException(400, { message: "viewer attach requires a live plane", }); } if (wantDesktop) requirePermission(grant, "stream:view"); if (wantTerminal) requirePermission(grant, "terminal:attach"); if (wantFiles) requirePermission(grant, "files:write"); // Consent gate (P3.2 / addendum E.1): only the explicitly requested, // un-redacted desktop plane needs acknowledgment. Terminal and files retain // their independent permission boundaries. if (wantDesktop) { const { shared } = await resolveSharedExposure(workspaceId, session); const ack = await getStreamAcknowledgment(db, { workspaceId, sandboxGroupId: session.sandboxGroupId, subjectId: grant.subjectId, }); if (!ack?.acknowledgedUnredacted) { throw new HTTPException(409, { message: "stream_acknowledgment_required", }); } if (shared && !ack.acknowledgedShared) { throw new HTTPException(409, { message: "shared_acknowledgment_required", }); } } // SELFHOSTED ACTIVE: when the session's active sandbox is selfhosted, skip // attachViewer (it warms the Modal group box — the wrong target). Synthesize a // result shaped like ViewerAttachResult and mint relay cells directly. const activeSandbox = session.activeSandboxId ? await getSandbox( db, { accountId: grant.accountId, workspaceId, subjectId: grant.subjectId }, session.activeSandboxId, ) : null; const selfhostedActive = activeSandbox?.kind === "selfhosted"; let stream: DesktopStreamMint | null = null; let terminal: TerminalStreamMint | null = null; let result: Awaited>; if (selfhostedActive) { const viewerId = parsed.data.viewerId ?? crypto.randomUUID(); result = { viewerId, liveness: "warm", leaseEpoch: session.activeEpoch, workspaceGeneration: null, archiveGeneration: null, archiveComplete: false, sandboxGroupId: session.sandboxGroupId, viewerHeartbeatIntervalMs: viewerHeartbeatIntervalMs(settings), dataPlaneUrl: null, }; if ( ((wantDesktop && settings.sandboxDesktopEnabled) || (wantTerminal && settings.sandboxTerminalEnabled)) && !streamTokenDegraded(settings) ) { if (wantDesktop && settings.sandboxDesktopEnabled) { stream = await mintDesktopStream(viewerServices, { accountId: grant.accountId, workspaceId, resourceSubjectId: grant.subjectId, resourceSubjectDelegated: grant.metadata?.delegated === true, session, viewerId, // No Modal lease for selfhosted-active; the mint routes to the relay. }); } if (wantTerminal && settings.sandboxTerminalEnabled) { terminal = await mintTerminalStream(viewerServices, { accountId: grant.accountId, workspaceId, resourceSubjectId: grant.subjectId, resourceSubjectDelegated: grant.metadata?.delegated === true, session, viewerId, // No Modal lease for selfhosted-active; the mint routes to the relay. }); } } } else { result = await attachViewer(viewerServices, { accountId: grant.accountId, workspaceId, session, viewerSubjectId: grant.subjectId, waitSignal: c.req.raw.signal, ...(parsed.data.viewerId ? { viewerId: parsed.data.viewerId } : {}), }); // P4.2 — the viewer now holds a WARM box; mint the real pixel cell IN-PROCESS // (resume by id → ensureDisplayStack → exposeStreamPort) scoped to THIS // viewer holder, record data_plane_url, and fold the live address into the // response. A degraded mint (no secret / headless / display-stack or tunnel // failure) leaves dataPlaneUrl null — the client falls back to Channel-A. The // box is warm here (attachViewer spun it up or attached), so the handshake's // never-spin-up rule does not apply. if ( ((wantDesktop && settings.sandboxDesktopEnabled) || (wantTerminal && settings.sandboxTerminalEnabled)) && !streamTokenDegraded(settings) ) { const lease = await readGroupLease( { db, settings }, { workspaceId, sandboxGroupId: session.sandboxGroupId }, ); if (lease) { // Mint only explicitly authorized plane credentials. The shared holder // supplies liveness; it is not itself authority for another plane. if (wantDesktop && settings.sandboxDesktopEnabled) { stream = await mintDesktopStream(viewerServices, { accountId: grant.accountId, workspaceId, resourceSubjectId: grant.subjectId, resourceSubjectDelegated: grant.metadata?.delegated === true, session, viewerId: result.viewerId, lease, }); } if (wantTerminal && settings.sandboxTerminalEnabled) { terminal = await mintTerminalStream(viewerServices, { accountId: grant.accountId, workspaceId, resourceSubjectId: grant.subjectId, resourceSubjectDelegated: grant.metadata?.delegated === true, session, viewerId: result.viewerId, lease, }); } } } } const response = { ...result, dataPlaneUrl: stream?.url ?? null, streamToken: stream?.token ?? null, streamExpiresAt: stream?.expiresAt ?? null, resolution: stream?.resolution ?? null, // Transport MUST match where the pixels were minted: a selfhosted-active box // serves the RELAY framebuffer (relay-frames/frames), a Modal box serves noVNC // (vnc-ws/novnc). Hardcoding vnc-ws here handed a machine's relay URL to the // noVNC renderer (and vice-versa on the swap-away case) → "closed before it // opened". Key off the SAME selfhostedActive the mint routed on. transport: stream ? selfhostedActive ? ("relay-frames" as const) : ("vnc-ws" as const) : null, client: stream ? (selfhostedActive ? ("frames" as const) : ("novnc" as const)) : null, // The REAL PTY terminal address (pty-ws), null when degraded. terminalUrl: terminal?.url ?? null, terminalToken: terminal?.token ?? null, terminalExpiresAt: terminal?.expiresAt ?? null, terminalTransport: terminal?.transport ?? null, } satisfies AttachViewerResponse; return c.json(response, 201); }); // POST .../viewers/:viewerId/heartbeat — refresh the holder TTL (epoch-fenced). // A holder may belong to any exact live plane. Lifecycle control accepts any // permission that could have minted it; the unguessable holder id and workspace // grant remain the ownership boundary. app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/viewers/:viewerId/heartbeat", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId); requireViewerLifecyclePermission(grant); assertOwnershipEnabled(); const sessionId = c.req.param("sessionId"); const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } const parsed = ViewerHeartbeatRequest.safeParse(await c.req.json().catch(() => ({}))); if (!parsed.success) { throw new HTTPException(400, { message: "viewer heartbeat requires { leaseEpoch }", }); } const alive = await heartbeatViewer( { db, settings }, { accountId: grant.accountId, workspaceId, sandboxGroupId: session.sandboxGroupId, viewerId: c.req.param("viewerId"), expectedEpoch: parsed.data.leaseEpoch, }, ); return c.json({ alive }); }, ); // DELETE .../viewers/:viewerId — release the holder (idempotent). app.delete("/v1/workspaces/:workspaceId/sessions/:sessionId/viewers/:viewerId", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId); requireViewerLifecyclePermission(grant); assertOwnershipEnabled(); const sessionId = c.req.param("sessionId"); const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } await detachViewer( { db, settings }, { accountId: grant.accountId, workspaceId, sandboxGroupId: session.sandboxGroupId, viewerId: c.req.param("viewerId"), }, ); return c.body(null, 204); }); // POST .../viewers/:viewerId/revoke — OD-6 v1 revocation. Drops the named // viewer's holder from the GROUP lease so refcount recomputes; the box drains // iff nothing else holds it (a turn-held or other-viewer-held box survives — // group-refcount liveness). Gated on stream:view (no new permission). The // live-RFB force-disconnect of an already-open socket is a P4 follow-up; the // holder-drop (so the box can drain) is the v1 deliverable. app.post( "/v1/workspaces/:workspaceId/sessions/:sessionId/viewers/:viewerId/revoke", async (c) => { const workspaceId = c.req.param("workspaceId"); const grant = await requireAccessGrant(c, deps, workspaceId, "stream:view"); assertOwnershipEnabled(); const sessionId = c.req.param("sessionId"); const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } const result = await revokeViewer(db, { accountId: grant.accountId, workspaceId, sandboxGroupId: session.sandboxGroupId, viewerId: c.req.param("viewerId"), idleGraceMs: settings.sandboxIdleGraceMs, }); // null ⇒ the lease was already cold-and-reaped (revoke is an idempotent no-op). return c.json({ liveness: result?.liveness ?? null, refcount: result?.refcount ?? null, }); }, ); // ══════════════════════ Channel-A structured services (P4.4) ══════════════ // // FileSystem (list/read/write/delete) + Git (status/diff/log/show) + Terminal // (exec + interactive PTY), all served API-DIRECT: each route does // requireAccessGrant BEFORE Zod parse -> resume the box by id in-process // (cold->warming CAS + viewer holder) -> SandboxChannelAService method // -> inline JSON -> release holder + drop handle. // NO Temporal, NO worker RPC, NO NATS round-trip — reads never ride the bus // (which would corrupt SSE gap-fill). The notifications (fs.changed/git.changed // /terminal.pty.*) ride A1 via appendAndPublishEvents. Gated behind // sandboxOwnershipEnabled (the lease is dormant otherwise). Explicit // HTTPException(400/404/409) — never a raw ZodError -> 500. // FS uses files:read for reads, files:write for mutations; Git is read-only // (rides files:read); Terminal exec + PTY ride terminal:attach. type ChannelARouteCtx = ChannelAContext & { grant: AccessGrant; waitSignal: AbortSignal; operation: ChannelAOperation; }; // Shared preamble: grant BEFORE parse, ownership gate, session lookup. Returns // the resolved context the channel-a seam needs (session narrowed non-null). async function channelAPreamble( c: Context, permission: "files:read" | "files:write" | "terminal:attach", operation: ChannelAOperation, ): Promise { const workspaceId = c.req.param("workspaceId") ?? ""; const grant = await requireAccessGrant(c, deps, workspaceId, permission); assertOwnershipEnabled(); const sessionId = c.req.param("sessionId") ?? ""; const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } return { grant, accountId: grant.accountId, workspaceId, session, subjectId: grant.subjectId, waitSignal: c.req.raw.signal, operation, }; } async function parseChannelABody( c: Context, schema: { safeParse: (v: unknown) => { success: true; data: T } | { success: false }; }, ): Promise { const raw = await c.req.json().catch(() => undefined); const result = schema.safeParse(raw ?? {}); if (!result.success) { throw new HTTPException(400, { message: "invalid request body" }); } return result.data; } // ── FileSystem ────────────────────────────────────────────────────────── app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/fs/list", async (c) => { const ctx = await channelAPreamble(c, "files:read", "fs.list"); const req = await parseChannelABody(c, FsListRequest); const out = await withChannelARead(channelAServices, ctx, ({ service }) => service.fsList(req)); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/fs/list-batch", async (c) => { const ctx = await channelAPreamble(c, "files:read", "fs.list-batch"); const req = await parseChannelABody(c, FsListBatchRequest); const out = await withChannelARead(channelAServices, ctx, async ({ service }) => ({ results: await runConcurrentChannelAReads( req.requests.map((request) => async () => await service.fsList(request)), ), })); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/fs/read", async (c) => { const ctx = await channelAPreamble(c, "files:read", "fs.read"); const req = await parseChannelABody(c, FsReadRequest); const out = await withChannelARead(channelAServices, ctx, ({ service }) => service.fsRead(req)); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/artifacts/publish", async (c) => { const workspaceId = c.req.param("workspaceId"); const authorization = await requireAccessGrantAuthorization( c, deps, workspaceId, "files:upload", ); const ctx = await channelAPreamble(c, "files:read", "artifact.publish"); const request = await parseChannelABody(c, PublishSandboxFileArtifactRequest); return c.json( await publishSandboxFileArtifact(deps, { grant: ctx.grant, authorization, session: ctx.session, path: request.path, signal: ctx.waitSignal, }), ); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/fs/write", async (c) => { const ctx = await channelAPreamble(c, "files:write", "fs.write"); const req = await parseChannelABody(c, FsWriteRequest); const out = await withChannelA(channelAServices, ctx, ({ service }) => service.fsWrite(req)); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/fs/delete", async (c) => { const ctx = await channelAPreamble(c, "files:write", "fs.delete"); const req = await parseChannelABody(c, FsDeleteRequest); const out = await withChannelA(channelAServices, ctx, ({ service }) => service.fsDelete(req)); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/fs/move", async (c) => { const ctx = await channelAPreamble(c, "files:write", "fs.move"); const req = await parseChannelABody(c, FsMoveRequest); const out = await withChannelA(channelAServices, ctx, ({ service }) => service.fsMove(req)); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/fs/mkdir", async (c) => { const ctx = await channelAPreamble(c, "files:write", "fs.mkdir"); const req = await parseChannelABody(c, FsMkdirRequest); const out = await withChannelA(channelAServices, ctx, ({ service }) => service.fsMkdir(req)); return c.json(out); }); // ── Git (read-only) ───────────────────────────────────────────────────── app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/git/status", async (c) => { const ctx = await channelAPreamble(c, "files:read", "git.status"); const req = await parseChannelABody(c, GitStatusRequest); const out = await withChannelARead(channelAServices, ctx, ({ service }) => service.gitStatus(req), ); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/git/diff", async (c) => { const ctx = await channelAPreamble(c, "files:read", "git.diff"); const req = await parseChannelABody(c, GitDiffRequest); const out = await withChannelARead(channelAServices, ctx, ({ service }) => service.gitDiff(req), ); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/git/read-batch", async (c) => { const ctx = await channelAPreamble(c, "files:read", "git.read-batch"); const req = await parseChannelABody(c, GitReadBatchRequest); const out = await withChannelARead(channelAServices, ctx, async ({ service }) => { type StatusResult = Awaited>; type DiffResult = Awaited>; type ReadResult = | { requestIndex: number; kind: "status"; value: StatusResult } | { requestIndex: number; kind: "diff"; value: DiffResult }; const operations: Array<() => Promise> = []; req.requests.forEach((request, requestIndex) => { operations.push(async () => ({ requestIndex, kind: "status" as const, value: await service.gitStatus(request.status), })); if (request.diff) { const diffRequest = request.diff; operations.push(async () => ({ requestIndex, kind: "diff" as const, value: await service.gitDiff(diffRequest), })); } }); const reads = await runConcurrentChannelAReads(operations); const statuses = new Map(); const diffs = new Map(); for (const read of reads) { if (read.kind === "status") statuses.set(read.requestIndex, read.value); else diffs.set(read.requestIndex, read.value); } return { results: req.requests.map((request, requestIndex) => { const status = statuses.get(requestIndex); if (!status) { throw new Error(`Workspace Git batch omitted status result ${requestIndex}.`); } const diff = request.diff ? diffs.get(requestIndex) : undefined; if (request.diff && !diff) { throw new Error(`Workspace Git batch omitted diff result ${requestIndex}.`); } return { status, ...(diff ? { diff } : {}) }; }), }; }); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/git/log", async (c) => { const ctx = await channelAPreamble(c, "files:read", "git.log"); const req = await parseChannelABody(c, GitLogRequest); const out = await withChannelARead(channelAServices, ctx, ({ service }) => service.gitLog(req)); return c.json(out); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/git/show", async (c) => { const ctx = await channelAPreamble(c, "files:read", "git.show"); const req = await parseChannelABody(c, GitShowRequest); const out = await withChannelARead(channelAServices, ctx, ({ service }) => service.gitShow(req), ); return c.json(out); }); // ── Workspace capture (read-only; served from DB + object storage, NO box) ── // Grant-first (files:read) then a pure DB/storage read — deliberately NOT the // channelAPreamble path: a capture is the durable turn-end snapshot, served // without warming a machine (the <200ms cold paint). No ownership-flag gate: // absent captures return {available:false} (200) so the client falls back to // the live/wake path — the feature degrades to today's behavior, never worse. app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/workspace/capture", async (c) => { const workspaceId = c.req.param("workspaceId") ?? ""; const grant = await requireAccessGrant(c, deps, workspaceId, "files:read"); const sessionId = c.req.param("sessionId") ?? ""; const lookup = await sessionLatestWorkspaceCapture(db, workspaceId, sessionId); if (!lookup.sessionExists) { throw new HTTPException(404, { message: "session not found" }); } if (!objectStorage) { // No storage configured → no captures can exist. Cold-fallback, not an error. return c.json({ available: false }); } return c.json( await serveWorkspaceCapture( lookup.capture, objectStorage, workspaceCaptureManifestCache, (fact) => recordAuditEvent(db, { accountId: grant.accountId, workspaceId, subjectId: grant.subjectId, action: "file.signed_url.issued", targetType: "workspace_capture", targetId: sessionId, metadata: { sessionId, ...fact }, }), ), ); }); app.get("/v1/workspaces/:workspaceId/sessions/:sessionId/workspace/capture/file", async (c) => { const workspaceId = c.req.param("workspaceId") ?? ""; const grant = await requireAccessGrant(c, deps, workspaceId, "files:read"); const sessionId = c.req.param("sessionId") ?? ""; const path = c.req.query("path"); if (!path) { throw new HTTPException(400, { message: "path query parameter is required", }); } const session = await getSession(db, workspaceId, sessionId); if (!session) { throw new HTTPException(404, { message: "session not found" }); } if (!objectStorage) { throw new HTTPException(404, { message: "capture not found" }); } // Explicit ?revision pins a specific capture; omitted → latest. const revisionParam = c.req.query("revision"); let row; if (revisionParam !== undefined && revisionParam !== "") { const revision = Number(revisionParam); if (!Number.isInteger(revision) || revision < 0) { throw new HTTPException(400, { message: "revision must be a non-negative integer", }); } row = await workspaceCaptureAtRevision(db, workspaceId, sessionId, revision); } else { row = await latestWorkspaceCapture(db, workspaceId, sessionId); } return c.json( await serveWorkspaceCaptureFile(row, path, objectStorage, (fact) => recordAuditEvent(db, { accountId: grant.accountId, workspaceId, subjectId: grant.subjectId, action: "file.signed_url.issued", targetType: "workspace_capture", targetId: sessionId, // The captured PATH is workspace-file identity (like a filename), // not content; the signed URL and object key are never recorded. metadata: { sessionId, path, ...fact }, }), ), ); }); // ── Terminal: synchronous exec ──────────────────────────────────────────── app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/terminal/exec", async (c) => { const ctx = await channelAPreamble(c, "terminal:attach", "terminal.exec"); const req = await parseChannelABody(c, TerminalExecRequest); const out = await withChannelA(channelAServices, ctx, ({ service }) => service.terminalExec(req), ); return c.json(out); }); // ── Terminal: interactive PTY control (output rides A1) ─────────────────── app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/terminal/pty", async (c) => { const ctx = await channelAPreamble(c, "terminal:attach", "terminal.pty.open"); const req = await parseChannelABody(c, PtyOpenRequest); if (ctx.session.sandboxBackend === "selfhosted" || ctx.session.activeSandboxId !== null) { throw new HTTPException(409, { message: "durable interactive terminals require the session-home provider route and are unavailable on active swaps or non-persistable routes; use synchronous exec or attach the session home sandbox", }); } const ptyId = crypto.randomUUID(); const out = await withChannelA(channelAServices, ctx, async (handle) => { if (!handle.lease) { throw new HTTPException(409, { message: "durable interactive terminals require a session-home provider lease", }); } const { service } = handle; const opened = await service.ptyOpen(req, ptyId); const execSessionId = opened.execSessionId; const retained = execSessionId === null ? null : handle.routingSession.retainedProcessIdentity(execSessionId); const process = retained ? await getRetainedProcess(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, processId: retained.id, }) : null; if ( execSessionId === null || !retained || !process || process.state !== "active" || process.ownerActorKind !== "direct" || process.providerSessionId !== execSessionId || process.routeTargetId !== null || process.leaseId !== handle.lease.id || process.sandboxGroupId !== handle.lease.sandboxGroupId || process.leaseEpoch !== handle.lease.leaseEpoch || process.providerBackend !== handle.lease.backend || process.providerInstanceId !== handle.lease.instanceId ) { if (execSessionId !== null && handle.routingSession.hasRetainedProcess(execSessionId)) { await drainOpenedPty(handle, execSessionId); } throw new HTTPException(409, { message: "interactive terminal did not acquire durable process authority", }); } const identity: SandboxPtyProcessIdentity = { leaseId: process.leaseId, sandboxGroupId: process.sandboxGroupId, retainedProcessId: process.id, openAdmissionId: process.parentAdmissionId, execSessionId: process.providerSessionId, leaseEpoch: process.leaseEpoch, providerBackend: process.providerBackend, providerInstanceId: process.providerInstanceId, routeKind: process.routeKind, routeTargetId: process.routeTargetId, routeEpoch: process.routeEpoch, }; try { await insertPtySession(db, { id: ptyId, accountId: ctx.accountId, workspaceId: ctx.workspaceId, sessionId: ctx.session.id, identity, cols: req.cols, rows: req.rows, shell: opened.shell, cwd: req.cwd, openedBy: ctx.subjectId, }); } catch (persistenceError) { try { await drainOpenedPty(handle, execSessionId); } catch (drainError) { failPtyPersistenceAndDrain(persistenceError, drainError); } throw persistenceError; } // Emit terminal.pty.started + any initial banner output on A1. const started: TerminalPtyStartedPayload = { ptyId, cols: req.cols, rows: req.rows, shell: opened.shell, cwd: req.cwd, }; const events: AppendEventInput[] = [{ type: "terminal.pty.started", payload: started }]; if (opened.initialOutput) { const delta: TerminalPtyOutputDeltaPayload = { ptyId, stream: "stdout", chunk: opened.initialOutput, seq: 0, }; events.push({ type: "terminal.pty.output.delta", payload: delta }); } await appendAndPublishEvents(db, bus, ctx.workspaceId, ctx.session.id, events); return opened.response; }); return c.json(out, 201); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/terminal/pty/write", async (c) => { const ctx = await channelAPreamble(c, "terminal:attach", "terminal.pty.write"); const req = await parseChannelABody(c, PtyWriteRequest); const pty = await getOpenPtySession(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, ptyId: req.ptyId, }); if (!pty) { throw new HTTPException(404, { message: "pty not found or closed" }); } let seq = 1; await withChannelA(channelAServices, ctx, async (handle) => { await adoptPtyProcess(ctx, handle, pty); let output: string; try { output = await handle.service.ptyWrite(req, pty.execSessionId, req.data); } catch (error) { const terminal = await getRetainedProcess(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, processId: pty.retainedProcessId, }); if (terminal && terminal.state !== "active") { await emitPtyExited(ctx, req.ptyId, terminal); } throw error; } const updated = await updatePtySessionActivity(db, { accountId: ctx.accountId, workspaceId: ctx.workspaceId, sessionId: ctx.session.id, ptyId: req.ptyId, identity: ptyIdentity(pty), }); if (!updated) { const terminal = await getRetainedProcess(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, processId: pty.retainedProcessId, }); if (terminal && terminal.state !== "active") { await emitPtyExited(ctx, req.ptyId, terminal); return; } throw new HTTPException(409, { message: "pty identity changed while input was in flight; reopen the terminal", }); } if (output) { const delta: TerminalPtyOutputDeltaPayload = { ptyId: req.ptyId, stream: "stdout", chunk: output, seq: seq++, }; await appendAndPublishEvents(db, bus, ctx.workspaceId, ctx.session.id, [ { type: "terminal.pty.output.delta", payload: delta }, ]); } }); return c.body(null, 204); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/terminal/pty/resize", async (c) => { const ctx = await channelAPreamble(c, "terminal:attach", "terminal.pty.resize"); const req = await parseChannelABody(c, PtyResizeRequest); const pty = await getOpenPtySession(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, ptyId: req.ptyId, }); if (!pty) { throw new HTTPException(404, { message: "pty not found or closed" }); } await withChannelA(channelAServices, ctx, async (handle) => { await adoptPtyProcess(ctx, handle, pty); await handle.service.ptyResize(req, pty.execSessionId); const updated = await updatePtySessionActivity(db, { accountId: ctx.accountId, workspaceId: ctx.workspaceId, sessionId: ctx.session.id, ptyId: req.ptyId, identity: ptyIdentity(pty), cols: req.cols, rows: req.rows, }); if (!updated) { throw new HTTPException(409, { message: "pty identity changed while resize was in flight; reopen the terminal", }); } }); return c.body(null, 204); }); app.post("/v1/workspaces/:workspaceId/sessions/:sessionId/terminal/pty/close", async (c) => { const ctx = await channelAPreamble(c, "terminal:attach", "terminal.pty.close"); const req = await parseChannelABody(c, PtyCloseRequest); const pty = await getOpenPtySession(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, ptyId: req.ptyId, }); // Idempotent: closing an already-closed/absent PTY is a 204 no-op. if (pty) { await withChannelA(channelAServices, ctx, async (handle) => { await adoptPtyProcess(ctx, handle, pty); await handle.service.ptyClose(req, pty.execSessionId); const terminal = await getRetainedProcess(db, { workspaceId: ctx.workspaceId, sessionId: ctx.session.id, processId: pty.retainedProcessId, }); if (!terminal || terminal.state === "active") { throw new HTTPException(409, { message: "pty close is pending exact provider exit proof; retry", }); } await emitPtyExited(ctx, req.ptyId, terminal); }); } return c.body(null, 204); }); } export function goalProposalMatchesExpectedRevision( revision: Pick, expectedObjectiveRevision: number, ): boolean { return revision.baseObjectiveRevision === expectedObjectiveRevision; } function eventListLimit(raw: string | undefined, max = 2000, fallback = 500): number { const limit = Number(raw ?? fallback); if (!Number.isFinite(limit)) { return fallback; } return Math.min(max, Math.max(1, Math.floor(limit))); } function codexRealtimeHttpFailure(error: CodexRealtimeBrokerError): { status: ContentfulStatusCode; code: ErrorCode; retryable: boolean; } { switch (error.reason) { case "invalid_request": case "incompatible": return { status: 422, code: "validation_failed", retryable: false }; case "entitlement_denied": return { status: 403, code: "forbidden", retryable: false }; case "rate_limited": return { status: 429, code: "limit_exceeded", retryable: true }; case "timeout": return { status: 504, code: "upstream_unavailable", retryable: true }; case "cancelled": return { status: 408, code: "upstream_unavailable", retryable: true }; case "provider_error": case "invalid_provider_response": case "network_error": return { status: 502, code: "upstream_unavailable", retryable: true }; case "subscription_disabled": case "credential_unavailable": case "reconnect_required": return { status: 409, code: "conflict", retryable: false }; } } function sessionRealtimeHttpError(error: unknown): HTTPException { if (error instanceof HTTPException) return error; if (error instanceof SessionRealtimeConflictError) { return new HTTPException(error.code === "REALTIME_NOT_FOUND" ? 404 : 409, { message: error.message, cause: error, }); } throw error; } /** * Map every mounted session-addressed HTTP path to the host-neutral operation * the embedding port authorizes. Returning null is deliberately fail-closed in * host-managed mode; standalone deployments never consult this classifier. */ export function sessionAuthorizationOperationForHttp( method: string, pathname: string, sessionId: string, ): SessionAuthorizationOperation | null { const marker = `/sessions/${sessionId}`; const markerAt = pathname.indexOf(marker); if (markerAt < 0) return null; const suffix = pathname.slice(markerAt + marker.length); const verb = method.toUpperCase(); if (suffix === "") { if (verb === "GET") return "session.read"; if (verb === "PATCH") return "session.title.write"; if (verb === "DELETE") return "session.delete"; return null; } if (suffix === "/pin" && verb === "PUT") return "session.pin.write"; if (suffix === "/attention" && verb === "PUT") return "session.attention.write"; if (suffix === "/archive" && verb === "PUT") return "session.archive.write"; if (suffix === "/visibility" && verb === "PUT") return "session.visibility.write"; if (suffix === "/forks" && verb === "POST") return "session.fork.create"; if (suffix === "/channel" && verb === "PUT") return "session.channel.write"; if (suffix === "/variable-sets" && verb === "PUT") return "session.variable_sets.write"; if (suffix === "/tool-policy" && verb === "PUT") return "session.tool_policy.write"; if (suffix === "/mcp-credentials/rotate" && verb === "POST") return "session.mcp.credentials.rotate"; if (/^\/mcp-servers\/[^/]+\/approval-policy$/.test(suffix) && verb === "PATCH") { return "session.mcp.approval_policy.write"; } if (suffix === "/lineage" && verb === "GET") return "session.lineage.read"; if (suffix === "/background-commands" && verb === "GET") return "session.read"; if (suffix === "/model-context" && verb === "GET") return "session.read"; if (/^\/background-commands\/[^/]+$/.test(suffix) && verb === "DELETE") { return "session.control"; } if (suffix === "/codex-account" && verb === "POST") { return "session.codex_account.write"; } if (suffix === "/realtime/webrtc" && verb === "POST") { return "session.realtime.start"; } if (suffix === "/realtime/gateway" && verb === "POST") { return "session.realtime.start"; } if (suffix === "/realtime/supergrok" && verb === "POST") { return "session.realtime.start"; } if (suffix === "/realtime" && verb === "POST") { return "session.realtime.start"; } if (/^\/realtime\/[^/]+\/heartbeat$/.test(suffix) && verb === "PATCH") { return "session.realtime.control"; } if (/^\/realtime\/[^/]+\/sync$/.test(suffix) && verb === "POST") { return "session.realtime.control"; } if (/^\/realtime\/[^/]+\/connections\/[^/]+\/activate$/.test(suffix) && verb === "POST") { return "session.realtime.control"; } if (/^\/realtime\/[^/]+$/.test(suffix) && verb === "DELETE") { return "session.realtime.control"; } if (suffix === "/goal") { return verb === "GET" ? "session.goal.read" : ["PATCH", "DELETE"].includes(verb) ? "session.goal.write" : null; } if (suffix === "/goal/revisions" && verb === "GET") return "session.goal.read"; if (suffix === "/goal/revisions/page" && verb === "GET") return "session.goal.read"; if (/^\/goal\/revisions\/[^/]+\/apply$/.test(suffix) && verb === "POST") { return "session.goal.write"; } if (/^\/goal\/revisions\/[^/]+\/(reject|rollback)$/.test(suffix) && verb === "POST") { return "session.goal.write"; } if (suffix === "/context/clear" || suffix === "/context/compact") { return verb === "POST" ? "session.context.write" : null; } if (suffix === "/events/stream" && verb === "GET") return "session.stream.read"; if (suffix === "/events") { if (verb === "GET") return "session.events.read"; if (verb === "POST") return "session.append"; return null; } if (suffix === "/turns" && verb === "GET") return "session.turns.read"; if (suffix === "/queue" && verb === "GET") return "session.queue.read"; if (suffix.startsWith("/queue/") && verb === "POST") return "session.queue.control"; if (suffix === "/composer-draft") { if (verb === "GET") return "session.composer.read"; if (verb === "PUT") return "session.composer.write"; return null; } // Same HTTP gate as POST /events. Delivery `steer` is re-authorized inside // acceptSessionUserMessageWithOutcome after the body is parsed. if (suffix === "/composer-draft/submit" && verb === "POST") return "session.append"; if (suffix === "/control" && verb === "POST") return "session.control"; if (suffix === "/steer" && verb === "POST") return "session.steer"; if (suffix === "/human-input-requests" && verb === "GET") { return "session.human_input.read"; } if (suffix.startsWith("/human-input-requests/") && verb === "GET") { return "session.human_input.read"; } if (suffix === "/stream-capabilities" && verb === "GET") return "session.viewer.read"; if (suffix === "/stream-capabilities/acknowledge" && verb === "POST") { return "session.stream.acknowledge"; } if (suffix === "/viewers" && verb === "POST") return "session.viewer.control"; if (suffix.startsWith("/viewers/") && ["POST", "DELETE"].includes(verb)) { return "session.viewer.control"; } if (suffix === "/fs/list" || suffix === "/fs/list-batch" || suffix === "/fs/read") { return verb === "POST" ? "session.files.read" : null; } if (suffix === "/artifacts/publish" && verb === "POST") return "session.files.write"; if (["/fs/write", "/fs/delete", "/fs/move", "/fs/mkdir"].includes(suffix)) { return verb === "POST" ? "session.files.write" : null; } if (suffix.startsWith("/git/") && verb === "POST") return "session.git.read"; if ((suffix === "/workspace/capture" || suffix === "/workspace/capture/file") && verb === "GET") { return "session.capture.read"; } if (suffix === "/terminal/exec" && verb === "POST") return "session.terminal.control"; if (suffix === "/terminal/pty" && verb === "POST") return "session.terminal.control"; if (suffix.startsWith("/terminal/pty/") && verb === "POST") { return "session.terminal.control"; } return null; } function sessionAuthorizationHttpError(error: unknown): HTTPException { if (error instanceof SessionAuthorizationDeniedError) { return new HTTPException(404, { message: "session not found" }); } if (error instanceof SessionAuthorizationUnavailableError) { return new HTTPException(503, { message: "session authorization is unavailable", }); } if (error instanceof HTTPException) return error; throw error; } export function sessionTenancyHttpError(error: unknown): Error { if (error instanceof SessionTenancyManagedHumanRequiredError) { return new ApiHttpError(403, { code: "forbidden", message: "Session tenancy mutations require the owning managed-human session.", retryable: false, }); } if (error instanceof SessionTenancyNotActivatedError) { return new ApiHttpError(409, { code: "conflict", message: "Session tenancy is not activated for this organization.", retryable: false, details: { reason: "not_activated" }, }); } if (error instanceof SessionTenancyConflictError) { return new ApiHttpError(409, { code: error.reason === "operation_reuse" ? "idempotency_conflict" : "conflict", message: error.reason === "not_quiescent" ? "The session must be fully quiescent before this change." : error.reason === "authority_epoch" ? "The session authority changed before this operation committed." : "The idempotency key was already used with different input.", retryable: false, details: { reason: error.reason, ...(error.blocker ? { blocker: error.blocker } : {}), }, }); } if ( error instanceof SessionTenancyAccessError || error instanceof SessionAuthorizationDeniedError ) { return new ApiHttpError(404, { code: "not_found", message: "Session not found.", retryable: false, }); } if (error instanceof SessionTenancyInvalidRequestError) { return new ApiHttpError(422, { code: "validation_failed", message: "Invalid session tenancy request.", retryable: false, }); } if (error instanceof SessionTenancyPersistenceOutcomeUnknownError) { return new ApiHttpError(503, { code: "upstream_unavailable", message: "The session tenancy mutation outcome is unknown. Retry with the same idempotency key.", retryable: true, outcomeUnknown: true, }); } if (error instanceof SessionAuthorizationUnavailableError) { return sessionAuthorizationHttpError(error); } return error instanceof Error ? error : new Error("Unknown session tenancy error"); } function eventEnumValue( raw: string | undefined, schema: { safeParse(value: unknown): { success: boolean; data?: T } }, name: string, fallback: T, ): T; function eventEnumValue( raw: string | undefined, schema: { safeParse(value: unknown): { success: boolean; data?: T } }, name: string, fallback: undefined, ): T | undefined; function eventEnumValue( raw: string | undefined, schema: { safeParse(value: unknown): { success: boolean; data?: T } }, name: string, fallback: T | undefined, ): T | undefined { if (raw === undefined) return fallback; const parsed = schema.safeParse(raw); if (!parsed.success) { throw new HTTPException(400, { message: `${name} is invalid` }); } return parsed.data as T; } function eventEnumList( raw: string | undefined, schema: { safeParse(value: unknown): { success: boolean; data?: T } }, name: string, ): T[] { if (raw === undefined || raw.trim() === "") return []; const values = raw .split(",") .map((value) => value.trim()) .filter(Boolean); if (values.length > 100) { throw new HTTPException(400, { message: `${name} accepts at most 100 values`, }); } return values.map((value) => { const parsed = schema.safeParse(value); if (!parsed.success) { throw new HTTPException(400, { message: `${name} contains an invalid value`, }); } return parsed.data as T; }); } function sessionListQuery( query: Record, allowCursor = true, ): { originSiteId: string | undefined; limit: string | undefined; parentSessionId: string | null | undefined; cursor: ReturnType | undefined; search: string | undefined; pinsOnly: boolean; archivedOnly: boolean; channelId: string | null | undefined; createdBy: { kind: "subject" | "service"; subjectId: string } | undefined; updatedFrom: Date | undefined; updatedBefore: Date | undefined; createdFrom: Date | undefined; createdBefore: Date | undefined; scopeSubjectId: SessionScopeSubjectId | undefined; hasPageFilters: boolean; } { const parentSessionId = query.parentSessionId; const originSiteId = query.originSiteId; if (originSiteId !== undefined && !z.string().uuid().safeParse(originSiteId).success) throw new HTTPException(400, { message: "originSiteId must be a Site id" }); // "null" = roots only; a uuid = children of that session; anything else is // a client error (an unvalidated value would surface as a Postgres uuid cast // failure -> 500 rather than an honest 400). if ( parentSessionId !== undefined && parentSessionId !== "null" && !z.string().uuid().safeParse(parentSessionId).success ) { throw new HTTPException(400, { message: 'parentSessionId must be a session id or the literal "null"', }); } const rawCursor = allowCursor ? query.cursor : undefined; const cursor = rawCursor ? decodeSessionListCursor(rawCursor) : undefined; if (rawCursor && !cursor) { throw new HTTPException(400, { message: "cursor is invalid" }); } const search = query.search?.trim(); if (search && search.length > 200) { throw new HTTPException(400, { message: "search must be at most 200 characters", }); } if (query.pinsOnly !== undefined && query.pinsOnly !== "true") { throw new HTTPException(400, { message: 'pinsOnly must be the literal "true"', }); } const pinsOnly = query.pinsOnly === "true"; if (query.archivedOnly !== undefined && query.archivedOnly !== "true") { throw new HTTPException(400, { message: 'archivedOnly must be the literal "true"' }); } const archivedOnly = query.archivedOnly === "true"; const channelId = query.channelId; if ( channelId !== undefined && channelId !== "null" && !z.string().uuid().safeParse(channelId).success ) { throw new HTTPException(400, { message: 'channelId must be a channel id or the literal "null"', }); } const createdByKind = query.createdByKind; const createdBySubjectId = query.createdBySubjectId; if ((createdByKind === undefined) !== (createdBySubjectId === undefined)) { throw new HTTPException(400, { message: "createdByKind and createdBySubjectId must be supplied together", }); } if (createdByKind !== undefined && createdByKind !== "subject" && createdByKind !== "service") { throw new HTTPException(400, { message: 'createdByKind must be "subject" or "service"', }); } if ( createdBySubjectId !== undefined && (createdBySubjectId.trim().length < 1 || createdBySubjectId.length > 1_024) ) { throw new HTTPException(400, { message: "createdBySubjectId must be between 1 and 1024 characters", }); } const parseDateBound = (name: string): Date | undefined => { const raw = query[name]; if (raw === undefined) return undefined; const parsed = z.string().datetime({ offset: true }).safeParse(raw); if (!parsed.success) { throw new HTTPException(400, { message: `${name} must be an ISO-8601 timestamp` }); } // Date-backed filters must not silently round accepted microseconds to an // earlier boundary. Clients can use any ISO offset with at most 3 decimals. if (/\.\d{4,}/.test(parsed.data)) { throw new HTTPException(400, { message: `${name} supports at most millisecond precision` }); } return new Date(parsed.data); }; const updatedFrom = parseDateBound("updatedFrom"); const updatedBefore = parseDateBound("updatedBefore"); const createdFrom = parseDateBound("createdFrom"); const createdBefore = parseDateBound("createdBefore"); if (updatedFrom && updatedBefore && updatedFrom >= updatedBefore) { throw new HTTPException(400, { message: "updatedFrom must be earlier than updatedBefore" }); } if (createdFrom && createdBefore && createdFrom >= createdBefore) { throw new HTTPException(400, { message: "createdFrom must be earlier than createdBefore" }); } // The opaque end-user label filter is an exact pair: one half alone is a // client error rather than a silently unfiltered list. if (query.endUserSource !== undefined || query.endUserId !== undefined) { throw new HTTPException(400, { message: "Use canonical scopeSubjectId, not an external-user label", }); } let scopeSubjectId: SessionScopeSubjectId | undefined; if (query.scopeSubjectId !== undefined) { const parsedEndUser = SessionScopeSubjectId.safeParse(query.scopeSubjectId); if (!parsedEndUser.success) { throw new HTTPException(400, { message: "scopeSubjectId must be a canonical OpenGeni user subject", }); } scopeSubjectId = parsedEndUser.data; } const hasPageFilters = originSiteId !== undefined || channelId !== undefined || createdByKind !== undefined || updatedFrom !== undefined || updatedBefore !== undefined || createdFrom !== undefined || createdBefore !== undefined || scopeSubjectId !== undefined; if (pinsOnly && !allowCursor) { throw new HTTPException(400, { message: 'pinsOnly requires view="page"' }); } if (pinsOnly && (rawCursor || parentSessionId !== undefined || search || hasPageFilters)) { throw new HTTPException(400, { message: "pinsOnly cannot be combined with cursor, parentSessionId, search, or filters", }); } if (pinsOnly && archivedOnly) { throw new HTTPException(400, { message: "pinsOnly cannot be combined with archivedOnly" }); } return { limit: query.limit, originSiteId, parentSessionId: parentSessionId === undefined ? undefined : parentSessionId === "null" ? null : parentSessionId, cursor, search: search || undefined, pinsOnly, archivedOnly, channelId: channelId === undefined ? undefined : channelId === "null" ? null : channelId, createdBy: createdByKind && createdBySubjectId ? { kind: createdByKind, subjectId: createdBySubjectId } : undefined, updatedFrom, updatedBefore, createdFrom, createdBefore, scopeSubjectId, hasPageFilters, }; } export type AgentTopologyCursorEnvelope = { cursor: SessionDiscoveryCursor; parentSessionId: string | null | "all"; rootSessionId: string | null; query: string | null; statuses: SessionStatus[]; activeOnly: boolean; recentHours: number | null; subject: WorkClaimSubjectFilter | null; claimLimit: number | null; }; export function encodeAgentTopologyCursor(value: AgentTopologyCursorEnvelope): string { return Buffer.from(JSON.stringify({ v: 2, ...value }), "utf8").toString("base64url"); } function decodeAgentTopologyCursor(value: string): AgentTopologyCursorEnvelope { if (value.length > 2_048) { throw new HTTPException(400, { message: "agent topology cursor is invalid", }); } try { const decoded = JSON.parse(Buffer.from(value, "base64url").toString("utf8")) as unknown; const legacy = z .object({ v: z.literal(1), parentSessionId: z.string().uuid().nullable(), search: z.string().max(WORK_DISCOVERY_QUERY_MAX_CHARS).nullable(), cursor: z.object({ orderBy: z.literal("updatedAt"), sortRevision: z.string().max(64), sortAt: z.string().max(64), id: z.string().uuid(), snapshotAt: z.string().max(64), snapshotRevision: z.string().max(64), updatedAfter: z.null(), }), }) .safeParse(decoded); if (legacy.success) { if (legacy.data.search !== null) { throw new Error("legacy search cursor is not relevance-fenced"); } const cursor = { ...legacy.data.cursor, sortRank: null, filterHash: null, } satisfies SessionDiscoveryCursor; return { cursor, parentSessionId: legacy.data.parentSessionId, rootSessionId: null, query: null, statuses: [], activeOnly: false, recentHours: null, subject: null, claimLimit: null, }; } const parsed = z .object({ v: z.literal(2), parentSessionId: z.union([z.string().uuid(), z.literal("all"), z.null()]), rootSessionId: z.string().uuid().nullable(), query: z.string().max(WORK_DISCOVERY_QUERY_MAX_CHARS).nullable(), statuses: z .array( z.enum([ "queued", "running", "idle", "requires_action", "recovering", "waiting_capacity", "failed", "cancelled", ]), ) .max(8), activeOnly: z.boolean(), recentHours: z.number().int().positive().max(WORK_DISCOVERY_RECENT_HOURS_MAX).nullable(), subject: z .object({ namespace: z.string().min(1).max(WORK_CLAIM_NAMESPACE_MAX_BYTES), type: WorkClaimSubjectType, canonicalKey: z.string().min(1).max(WORK_CLAIM_CANONICAL_KEY_MAX_BYTES), }) .strict() .nullable(), claimLimit: z.number().int().positive().max(WORK_CLAIM_DISCOVERY_LIMIT).nullable(), cursor: z.object({ orderBy: z.enum(["updatedAt", "relevance"]), sortRank: z.number().int().nonnegative().nullable(), sortRevision: z.string().max(64), sortAt: z.string().max(64), id: z.string().uuid(), snapshotAt: z.string().max(64), snapshotRevision: z.string().max(64), updatedAfter: z.null(), filterHash: z .string() .regex(/^[0-9a-f]{64}$/) .nullable(), }), }) .parse(decoded); if ( !/^(?:0|[1-9]\d*)$/.test(parsed.cursor.sortRevision) || !/^(?:0|[1-9]\d*)$/.test(parsed.cursor.snapshotRevision) || BigInt(parsed.cursor.sortRevision) > 9_223_372_036_854_775_807n || BigInt(parsed.cursor.snapshotRevision) > 9_223_372_036_854_775_807n || Number.isNaN(Date.parse(parsed.cursor.sortAt)) || Number.isNaN(Date.parse(parsed.cursor.snapshotAt)) ) { throw new Error("invalid topology cursor fields"); } if ( (parsed.cursor.orderBy === "relevance" && (parsed.cursor.sortRank === null || parsed.cursor.filterHash === null)) || (parsed.cursor.orderBy === "updatedAt" && (parsed.cursor.sortRank !== null || parsed.cursor.filterHash !== null)) ) { throw new Error("invalid topology cursor relevance fields"); } return parsed; } catch { throw new HTTPException(400, { message: "agent topology cursor is invalid", }); } } export function agentTopologyQuery(query: Record): { limit: number; parentSessionId: string | null | undefined; rootSessionId: string | undefined; query: string | undefined; statuses: SessionStatus[] | undefined; activeOnly: boolean; recentHours: number | undefined; subject: WorkClaimSubjectFilter | undefined; claimLimit: number | undefined; cursor: SessionDiscoveryCursor | undefined; } { const rawLimit = query.limit; const limit = rawLimit === undefined ? 25 : Number(rawLimit); if (!Number.isInteger(limit) || limit < 1 || limit > 100) { throw new HTTPException(400, { message: "limit must be an integer between 1 and 100", }); } const rawParent = query.parentSessionId; if ( rawParent !== undefined && rawParent !== "null" && !z.string().uuid().safeParse(rawParent).success ) { throw new HTTPException(400, { message: 'parentSessionId must be a session id or the literal "null"', }); } const rootSessionId = query.rootSessionId?.trim(); if (rootSessionId && !z.string().uuid().safeParse(rootSessionId).success) { throw new HTTPException(400, { message: "rootSessionId must be a session id" }); } const normalizeSearchQuery = (value: string | undefined): string | undefined => { if (value === undefined) return undefined; const canonical = value.normalize("NFKC"); if (/[\u0000-\u001f\u007f-\u009f]/u.test(canonical)) { throw new HTTPException(400, { message: "query must not contain control characters" }); } const normalized = canonical.trim().replace(/\s+/gu, " ").toLowerCase(); if (!normalized) return undefined; if (Array.from(normalized).length > WORK_DISCOVERY_QUERY_MAX_CHARS) { throw new HTTPException(400, { message: `query must be at most ${WORK_DISCOVERY_QUERY_MAX_CHARS} characters`, }); } return normalized; }; const requestedQuery = normalizeSearchQuery(query.query); const legacySearch = normalizeSearchQuery(query.search); const searchQuery = requestedQuery ?? legacySearch; if (requestedQuery && legacySearch && requestedQuery !== legacySearch) { throw new HTTPException(400, { message: "query and legacy search must match" }); } const statuses = query.statuses ? [ ...new Set( query.statuses .split(",") .map((status) => status.trim()) .filter(Boolean), ), ].sort() : []; const parsedStatuses = z .array( z.enum([ "queued", "running", "idle", "requires_action", "recovering", "waiting_capacity", "failed", "cancelled", ]), ) .max(8) .safeParse(statuses); if (!parsedStatuses.success) { throw new HTTPException(400, { message: "statuses contains an unsupported lifecycle state" }); } const activeOnly = query.activeOnly === "true"; if ( query.activeOnly !== undefined && query.activeOnly !== "true" && query.activeOnly !== "false" ) { throw new HTTPException(400, { message: "activeOnly must be true or false" }); } const recentHours = query.recentHours === undefined ? undefined : Number(query.recentHours); if ( recentHours !== undefined && (!Number.isSafeInteger(recentHours) || recentHours < 1 || recentHours > WORK_DISCOVERY_RECENT_HOURS_MAX) ) { throw new HTTPException(400, { message: `recentHours must be an integer between 1 and ${WORK_DISCOVERY_RECENT_HOURS_MAX}`, }); } const subjectFields = [query.subjectNamespace, query.subjectType, query.subjectKey]; if (subjectFields.some((value) => value !== undefined) && subjectFields.some((value) => !value)) { throw new HTTPException(400, { message: "subjectNamespace, subjectType, and subjectKey must be supplied together", }); } const parsedSubject = subjectFields.every((value) => value !== undefined) ? WorkClaimSubjectFilterSchema.safeParse({ namespace: normalizeWorkClaimNamespace(query.subjectNamespace!), type: query.subjectType, canonicalKey: normalizeWorkClaimCanonicalKey(query.subjectKey!), }) : null; if (parsedSubject && !parsedSubject.success) { throw new HTTPException(400, { message: "exact subject filter is invalid" }); } const subject = parsedSubject?.success ? parsedSubject.data : undefined; if (searchQuery && subject) { throw new HTTPException(400, { message: "query cannot be combined with an exact subject" }); } const relevanceRequested = Boolean(searchQuery || subject); // An explicit root scopes the whole workstream; only an explicit parent // narrows it to one level. Ordinary browse still defaults to root sessions. const parentSessionId = rawParent === undefined ? relevanceRequested || Boolean(rootSessionId) ? undefined : null : rawParent === "null" ? null : rawParent; const claimLimit = query.claimLimit === undefined ? undefined : Number(query.claimLimit); if ( claimLimit !== undefined && (!Number.isSafeInteger(claimLimit) || claimLimit < 1 || claimLimit > WORK_CLAIM_DISCOVERY_LIMIT) ) { throw new HTTPException(400, { message: `claimLimit must be an integer between 1 and ${WORK_CLAIM_DISCOVERY_LIMIT}`, }); } const envelope = query.cursor ? decodeAgentTopologyCursor(query.cursor) : undefined; const expectedEnvelope = { parentSessionId: parentSessionId === undefined ? "all" : parentSessionId, rootSessionId: rootSessionId ?? null, query: searchQuery || null, statuses: parsedStatuses.data, activeOnly, recentHours: recentHours ?? null, subject: subject ?? null, claimLimit: claimLimit ?? null, }; if ( envelope && JSON.stringify({ parentSessionId: envelope.parentSessionId, rootSessionId: envelope.rootSessionId, query: envelope.query, statuses: envelope.statuses, activeOnly: envelope.activeOnly, recentHours: envelope.recentHours, subject: envelope.subject, claimLimit: envelope.claimLimit, }) !== JSON.stringify(expectedEnvelope) ) { throw new HTTPException(400, { message: "agent topology cursor does not match its filters", }); } return { limit, parentSessionId, rootSessionId: rootSessionId || undefined, query: searchQuery || undefined, statuses: parsedStatuses.data.length > 0 ? parsedStatuses.data : undefined, activeOnly, recentHours, subject, claimLimit, cursor: envelope?.cursor, }; } function compactEvents(raw: string | undefined): boolean { return raw === "1" || raw === "true"; } function eventSequence(raw: string | undefined, fallback: number): number { const sequence = Number(raw ?? fallback); if (!Number.isFinite(sequence)) { return fallback; } return Math.floor(sequence); } function optionalEventSequence(raw: string | undefined): number | undefined { if (raw === undefined) { return undefined; } const sequence = Number(raw); if (!Number.isFinite(sequence)) { return undefined; } return Math.floor(sequence); } /** Stable, value-free JSON errors for only the create-session boundary. */ export function sessionCreateErrorResponse(c: Context, error: unknown): Response { if (error instanceof SessionTenancyManagedHumanRequiredError) { return c.json( { code: "SESSION_CREATE_FORBIDDEN", message: "Only managed-account users can create an Only-me session.", }, 403, ); } if (error instanceof SessionTenancyNotActivatedError) { return c.json( { code: "SESSION_TENANCY_NOT_ACTIVATED", message: "Private sessions are not enabled for this organization.", }, 409, ); } if (error instanceof ChannelNotFoundError) { // Covers the create-vs-channel-delete race the pre-validation cannot: the // insert's FK rejection surfaces as the same 422 an unknown id gets. return c.json({ code: "SESSION_CREATE_REJECTED", message: error.message }, 422); } if (error instanceof SessionVariableSetSelectionUnavailableError) { return c.json( { code: "SESSION_CREATE_REJECTED", message: error.message, details: { variableSetIds: error.variableSetIds }, }, 422, ); } if (error instanceof SessionSpawnDeniedError) { return c.json( sessionSpawnDenialEnvelope(error), error.denial.code === "nested_agent_depth_override_forbidden" ? 403 : 409, ); } if (error instanceof ZodError) { return c.json( { code: "INVALID_SESSION_CREATE_REQUEST", message: `Invalid session create request: ${zodErrorFields(error)} failed schema validation`, }, 422, ); } if ( error instanceof HTTPException && error.status === 409 && error.cause instanceof NewSessionDraftConflictError ) { return c.json( { code: "NEW_SESSION_DRAFT_CONFLICT", message: error.cause.message, currentRevision: error.cause.currentRevision, }, 409, ); } if (error instanceof HTTPException && error.status === 422) { return c.json( { code: "SESSION_CREATE_REJECTED", message: error.message, }, 422, ); } if (error instanceof HTTPException && error.status === 409) { return c.json( { code: "SESSION_CREATE_CONFLICT", message: error.message, }, 409, ); } throw error; } export function parseSessionEventAdmission(raw: unknown): ClientSessionEvent { const parsed = ClientSessionEvent.safeParse(raw); if (!parsed.success) { throw new HTTPException(422, { message: "invalid session event" }); } return parsed.data; } export function parseSteerSessionAdmission(raw: unknown): SteerSessionMessageRequest { const parsed = SteerSessionMessageRequest.safeParse(raw); if (!parsed.success) { throw new HTTPException(422, { message: "invalid steer request" }); } return parsed.data; } function zodErrorFields(error: ZodError): string { const paths = [ ...new Set( error.issues.map((issue) => { const path = issue.path.map(String).join("."); return path || "request"; }), ), ]; const shown = paths.slice(0, 5); const remainder = paths.length - shown.length; return `${shown.join(", ")}${remainder > 0 ? `, and ${remainder} more` : ""}`; } function commandConflictResponse(c: Context, error: unknown): Response { if (error instanceof QueueCommandConflictError) { return c.json({ code: error.code, message: error.message, current: error.current }, 409); } if (error instanceof SessionControlConflictError) { return c.json({ code: error.code, message: error.message }, 409); } if (error instanceof SessionCommandIdempotencyError) { return c.json({ code: error.code, message: error.message }, 409); } throw error; } type EffectivePolicyContext = { workspaceServerIds: string[]; workspaceDefaultServerIds: string[]; }; async function loadEffectivePolicyContext( deps: ApiRouteDeps, workspaceId: string, subjectId: string, ): Promise { const [workspaceServerIds, workspaceDefaultServerIds] = await Promise.all([ workspaceSessionToolPolicyServerIds(deps.db, workspaceId, deps.settings, subjectId), workspaceSessionToolPolicyDefaultServerIds(deps.db, workspaceId, deps.settings, subjectId), ]); return { workspaceServerIds, workspaceDefaultServerIds }; } async function withEffectivePolicy( deps: ApiRouteDeps, workspaceId: string, subjectId: string, session: Session, ): Promise { const policy = await loadEffectivePolicyContext(deps, workspaceId, subjectId); return sessionWithEffectiveToolPolicy( session, policy.workspaceServerIds, policy.workspaceDefaultServerIds, ); } function mapLineageNodes(nodes: LineageNode[], policy: EffectivePolicyContext): LineageNode[] { return nodes.map((node) => ({ ...node, session: sessionWithEffectiveToolPolicy( node.session as Session, policy.workspaceServerIds, policy.workspaceDefaultServerIds, ), children: mapLineageNodes(node.children, policy), })); }