src/cloudflare.ts
Rama main · réplica de lectura del repositorio original.
import { GatewayInboxConflictError } from './slack/gateway/inbox.ts';
import { withAgentAppExecution } from './slack/agent-apps/index.ts';
import { GATEWAY_HTTP_SETTING, parseHttpDeliveryState, verifyHttpDelivery, httpDeliveryReceipt, HttpDeliveryError } from './slack/gateway/http-delivery.ts';
import { GATEWAY_BINDING_SETTING } from './slack/gateway/client.ts';
import type { GatewayAdmissionResult, GatewayInboundDelivery, GatewayWorkspaceBinding } from './slack/gateway/protocol.ts';
import type { AgentAppBot } from './slack/agent-app-bots.ts';
import { gatewayUiReceiptNow, isKeyFormDelivery } from './slack/gateway/ui-receipt.ts';
import { scheduleActionRpcResult } from './management/slack-schedule-rpc.ts';
import {
DurableObject,
env,
type DurableObjectState,
} from 'cloudflare:workers';
import { Sandbox as CloudflareSandbox } from '@cloudflare/sandbox';
import { instrument } from '@flue/runtime';
import { createCloudflareTracing } from '@flue/runtime/cloudflare';
import { emitManagementToolFailure } from './management/telemetry.ts';
import { emitRuntimeCorrelation } from './work/trace-correlation.ts';
import {
emitGatewayDelivery,
emitRelayAlarm,
emitRuntimeLatency,
startRelayAlarmMetrics,
StateRpcSummary,
StoreBootMarks,
type RelayAlarmMetrics,
} from './observability/runtime-latency.ts';
import {
AgentConnectionCredentialConflictError,
AgentRevisionConflictError,
AgentExistsError,
AgentStillAssignedError,
AgentStillReferencedError,
ChannelRevisionConflictError,
ConnectionAccountAlreadyBoundError,
ConnectionAccountRevisionConflictError,
ManagedRemoteAccountAlreadyUsedError,
ManagedSignInGroupConflictError,
ReservedAgentIdentityError,
UnknownAgentError,
ModelRoleRevisionConflictError,
WorkspaceModelDefaultRevisionConflictError,
} from './config/errors.ts';
import { slackAgentThreadKey } from './slack/thread-key.ts';
import { gitIdentityConfigCommand, resolveWorkspaceGitIdentity } from './sandbox/git-identity.ts';
import { recordDeliveredSlackAgentMessage } from './slack/public-context.ts';
import {
cacheSlackInstallationExecutionContexts,
carriedGatewayIdentity,
effectiveTurnSlackInstallationId,
resolveSlackInstallationExecutionContext,
verifySlackInstallationTurnAccess,
type SlackInstallationExecutionResolver,
} from './slack/installation-execution.ts';
import { SANDBOX_PACKAGE_REGISTRY_HOSTS } from './config/sandbox-settings.ts';
import type {
ModelCredentialStore,
PublishModelCredentialInput,
ReplaceEncryptedCredentialRevisionInput,
RewrapModelCredentialInput,
SettingsPatch,
SettingsStore,
} from './config/settings-store.ts';
import { purgeExpiredImageOutputs } from './images/output-store.ts';
import type {
StateRpcResult,
StateRpcErrorCode,
SlackWorkspaceManagementRpcRequest,
TagStateRpc,
TurnJob,
TurnProgress,
TurnPullRequestProgress,
RuntimeDrainStatus,
} from './config/state-rpc.ts';
import { buildRuntimeDrainStatus, tagStateStub } from './config/state-rpc.ts';
import type { StateRead } from './config/turn-reads.ts';
import {
deploymentTenancy,
installationOwnershipOf,
installationScopeOf,
objectInstallationEnv,
requireInstallationScope,
} from './config/installation-scope.ts';
import type { UiSurfaceRpcRequest } from './slack/ui/surface-store.ts';
import {
getConfigStore,
getIdentityStore,
getSettingsStore,
getSlackStateStore,
type AppStores,
type PlatformEnv,
} from './config/state-backend.ts';
import {
isOAuthContinuationActorActive,
repairPendingOAuthContinuationResumes,
} from './connections/oauth-continuation.ts';
import {
type AgentCreationFence,
type AgentModelRolePatch,
type ConfigAgentPatch,
type OAuthReauthorizationTarget,
type ApiCredentialTarget,
} from './config/store.ts';
import type {
ActivateChickpeaCutoverInput,
AgentCreateInput,
AgentChannelGrant,
AgentChannelGrantInput,
AgentModelRole,
AgentModelRoleInput,
AgentConnectionBinding,
AgentConnectionBindingInput,
AgentOwnedConnection,
AgentOwnedConnectionInput,
SignInGroupChange,
SignInReferenceRelease,
AgentScheduleReference,
AgentScheduleReferenceInput,
AgentSnapshot,
AgentSnapshotRootReference,
AgentReferenceSummary,
AgentThreadRoute,
AgentThreadRouteInput,
ChannelConfig,
ChickpeaCutoverActivation,
ChickpeaCutoverPreflight,
CustomAgentConfig,
ConnectionAccount,
ConnectionAccountInput,
EnsureWorkspaceInstallationInput,
PrepareChickpeaCutoverInput,
RollbackChickpeaCutoverInput,
SlackPublicContextEntry,
SlackPublicContextEntryInput,
RecentSlackAgentRepliesInput,
RecentSlackAgentReply,
RecentSlackPublicContextInput,
NonChatModelRole,
WorkspaceModelDefault,
WorkspaceModelDefaultInput,
WorkspaceModelRole,
WorkspaceModelRoleInput,
WorkspaceInstallation,
WorkspaceInstallationPatch,
} from './config/types.ts';
import {
SandboxPolicyState,
type SandboxEgressContext,
type SandboxEgressPolicy,
type SandboxEgressPolicyInput,
type SandboxPolicyStorage,
} from './sandbox/cloudflare-policy.ts';
import {
denySandboxOutbound,
githubSandboxOutbound,
packageRegistrySandboxOutbound,
type SandboxOutboundHandler,
} from './sandbox/egress-outbound.ts';
import { sandboxObjectEnv, sandboxTurnReaders, type SandboxObjectContext } from './sandbox/sandbox-object.ts';
import { containerStopRecorded, sandboxHostFunctions, type SandboxHostRpc } from './sandbox/sandbox-host.ts';
import type { SandboxContainerAdmission } from './sandbox/container-lease.ts';
import {
admitWorkspaceContainer,
holdStartedWorkspaceContainer,
isContainerLeaseSweepMinute,
maintainWorkspaceContainerLeases,
meterWorkspaceContainer,
refuseUnadmittedContainerStart,
requireWorkspaceTurnAdmitted,
} from './sandbox/hosted-limits.ts';
import type { SandboxContainerLimits } from './config/sandbox-settings.ts';
import { isCheckpointSweepMinute, sweepInstallationCheckpoints } from './sandbox/checkpoint-sweep.ts';
import { maintainWebsiteLogins } from './browser/logins.ts';
import { maintainBrowserSessionLeases } from './browser/session-lease.ts';
import { isOAuthKeepAliveMinute, runOAuthKeepAliveSweep } from './connections/oauth-keepalive.ts';
import {
SandboxWorkspaceState,
WORKSPACE_CHECKPOINT_EXCLUDES,
WORKSPACE_CHECKPOINT_TTL_SECONDS,
WORKSPACE_DIR,
type WorkspaceTurnState,
} from './sandbox/workspace-lifecycle.ts';
import {
readStoredWorkspaceRoster,
type WorkspaceRosterState,
} from './sandbox/workspace-limits.ts';
import {
putStoredCodingTask,
readStoredCodingTasks,
settleStoredCodingTask,
type CodingTaskRecordV1,
} from './sandbox/coding-task-record.ts';
import {
checkpointWorkspace,
discardWorkspaceCheckpoint,
restoreWorkspaceCheckpoint,
workspaceCheckpointsAvailable,
} from './sandbox/workspace-checkpoints.ts';
import {
clearWarmWindow,
holdWarmWindow,
stopPastWarmWindow,
type WarmWindowStorage,
workspaceWarmWindowMs,
} from './sandbox/warm-window.ts';
import type {
SlackCanonicalAdmissionInput,
SlackDeliveryNote,
SlackDirectDelivery,
SlackDirectDeliveryResult,
SlackFrontDeskHandoffNote,
SlackStateStore,
} from './slack/claim-store.ts';
import {
resolveSlackInstallationCredentials,
SlackCredentialRecoveryOnlyError,
type ResolvedSlackInstallationCredentials,
} from './slack/installation-credentials.ts';
import { recordFirstHostedSlackDelivery } from './slack/hosted-installation.ts';
import type { SlackEventFixture } from './slack/types.ts';
import { ConversationOrder } from './state/conversation-order.ts';
import type { SlackMessageEdit } from './slack/edits.ts';
import {
SlackPresentationStateError,
} from './slack/run-presentations.ts';
import { createLedgerSlackRunHandler } from './slack/ledger-turn-driver.ts';
import type { SlackPresentationStatePort } from './slack/agent-view-presentation.ts';
import { CfSlackStateStore } from './config/cf-state-proxies.ts';
import type { SlackPresentationTransitionInput, SlackReadMethod } from './slack/run-presentations.ts';
import { defaultSlackStatusRegistry } from './slack/status-registry.ts';
import {
activityStatus,
isSafeTypedActivityStatus,
type TypedActivityStatus,
} from './activity/status.ts';
import {
repairSlackInteractionProgress,
runTurn,
sanitizeError,
} from './slack/run-turn.ts';
import {
ALARM_ADMISSION_RECHECK_MS,
ALARM_PENDING_PER_THREAD,
ALARM_TURN_BUDGET_MS,
ALARM_TURN_HARD_CAP_MS,
ALARM_YIELD_REARM_MS,
drainAlarmTurnJobs,
type AlarmTurnJobControl,
} from './slack/alarm-turn-drain.ts';
import {
executeTurnJob,
type TurnExecutionPorts,
} from './slack/turn-executor.ts';
import { slackTurnExecutor } from './slack/turn-executor-flag.ts';
import {
boundedStopCall,
receiveAlarmExecutorStop,
STOP_NOTICE_DELIVERY_TIMEOUT_MS,
StopAbortFence,
} from './slack/runner-stops.ts';
import { abortSlackThreadAgent } from './slack/flue-dispatch.ts';
import {
runnerPrefetchedSettings,
threadRunnerStub,
type RunnerTurnBegin,
type SlackThreadRunnerRpc,
type ThreadRunnerJobPayload,
type ThreadRunnerTurnKind,
type ThreadRunnerTurnOp,
type ThreadRunnerTurnResult,
} from './slack/thread-runner-rpc.ts';
import { receiptRunnerWake, type ReceiptRunnerWake, type ReceiptWakeDelivery } from './slack/receipt-runner-wake.ts';
import { admissionAgentInstance, type AdmissionAgentFacts, type AdmissionAgentWake } from './slack/admission-agent-wake.ts';
import {
admissionStatusPresentation,
showAdmissionStatus,
type AdmissionStatusMark,
} from './slack/admission-activity.ts';
import { CHICKPEA_SLACK_AGENT_BINDING } from './slack/bounded-agent-observation.ts';
import { wakeSlackAgentObject } from './agents/turn-input.ts';
import { localSlackPresentationStatePort } from './slack/presentation-state-port.ts';
import {
drainSlackPresentationRepairs,
type SlackPresentationRepairDrainResult,
} from './slack/presentation-repair.ts';
import {
deliverDueStopNotices,
MAX_TURN_DRAIN_BATCH,
oauthResumeTurnJobId,
type PendingTurnJob,
} from './slack/turn-jobs.ts';
import type { TurnFoldRequest, TurnSteeringDecision } from './slack/turn-job-types.ts';
import { DoSqlStateDb } from './state/do-state-db.ts';
import {
buildTagStateStores,
localGatewayAppStores,
localTurnEnqueue,
type TagStateStores,
} from './state/tag-state-stores.ts';
import type {
HostObjectStorage,
ObjectExportRequest,
ObjectHostRequest,
ObjectRestoreBookmarksRequest,
ObjectRestoreRequest,
ObjectRestoreRestartRequest,
} from './state/object-host.ts';
import type { InstallationWorkspaceObject } from './state/object-inventory.ts';
import { ResidencyHold, residencyHoldMs } from './state/residency-hold.ts';
import { stateStoreHostFunctions, type StateStoreHostRpc } from './state/state-store-host.ts';
import type { TestRunLifecycle } from './state/test-run-lifecycle.ts';
import type {
PracticeRunListing,
PracticeRunsQuery,
TestRunQuery,
TestRunReplayQuery,
TestRunStartRequest,
TestRunStartResult,
TestRunView,
} from './management/test-run-types.ts';
import { StateSchemaMarker, stateSchemaFingerprint } from './state/schema-lifecycle.ts';
import { cloudflareWorkerVersionId } from './config/cloudflare-version.ts';
import { applicationIdentity, viteServeLane } from './release/identity.ts';
import { registerCloudflareBindingProvider } from './cloudflare-provider.ts';
import { MemoryStateError, type MemoryRpcRequest, type MemoryRpcResponse } from './memory/types.ts';
import {
RoutineStateError,
type RoutineRpcRequest,
type RoutineRpcResponse,
} from './routines/types.ts';
import {
createRoutineScheduledHandler,
runWithGuaranteedFinalizer,
} from './routines/scheduler-adapter.ts';
import {
GATEWAY_INBOX_MAX_DRAIN_BATCH,
gatewayDeliveryFailureReason,
gatewayDeliveryRetryDelayMs,
recordGatewayDeliveryDeadLetter,
} from './slack/gateway/inbox.ts';
import {
GatewayDeploymentClient,
} from './slack/gateway/client.ts';
import { resolveChickpeaGatewayUrl } from './slack/gateway/runtime.ts';
import { loadCredentialKeyring } from './slack/credential-keyring.ts';
import { slackInstallationCredentialId } from './slack/hosted-slack-app.ts';
import {
processGatewayAgentSelection,
processGatewayPrivateChannelSetup,
processGatewayUiAction,
processGatewayViewSubmission,
processGatewaySlackEnvelope,
processDirectSlackEvent,
processSlackAgentAsks,
processSlackFrontDeskHandoff,
admitSetupWake,
} from './channels/slack.ts';
import {
SlackGatewaySession,
wakeCloudflareGatewaySession,
} from './slack/gateway/cloudflare-session.ts';
import { UsageStateError } from './usage/store-error.ts';
import type { UsageRpcRequest, UsageRpcResponse, UsageStore } from './usage/types.ts';
import { IdentityStateError } from './identity/errors.ts';
import {
bindStoreInstallation,
type InstallationIdentity,
} from './identity/installation-binding.ts';
import type { IdentityStore } from './identity/types.ts';
import type { IdentityRpcRequest, IdentityRpcResponse } from './identity/types.ts';
import type { ManagementStore } from './management/store.ts';
import { createLiveWorkspaceManagementService } from './management/live-service.ts';
import { createPlatformProductTelemetry } from './telemetry/platform.ts';
import type { ProductTelemetryCapture } from './telemetry/client.ts';
import type { SlackAgentAskDispatcher } from './slack/agent-asks.ts';
import { frontDeskHandoffRunnerWake, type SlackFrontDeskHandoffDispatcher } from './slack/front-desk.ts';
import { createWaitUntilTelemetryLifecycle } from './telemetry/runtime.ts';
import {
invokeSlackWorkspaceManagementTool,
resolveSlackManagementActor,
} from './management/slack-tools.ts';
import {
executeHostSlackManagementApproval,
type HostSlackManagementApprovalResult,
type SlackManagementApprovalRpcRequest,
} from './management/slack-approval.ts';
import { agentAvatarInstallationField } from './slack/agent-presence/avatar-assets.ts';
import {
invokeSlackScheduleAction,
retryDueSlackScheduleActions,
type SlackScheduleActionOutcome,
type SlackScheduleActionRpcRequest,
} from './management/slack-schedule-actions.ts';
import type { WorkspaceManagementToolResult } from './management/tool-adapter.ts';
import {
completeAgentWelcomeDelivery,
completeSettledAgentWelcomeHandoff,
deliverManagementReceiptToSlack,
drainManagementReceiptOutbox,
failAgentWelcomeTurn,
isAgentCreatedWelcome,
reconcileScheduleActionReceipts,
} from './management/receipts.ts';
import {
ManagementError,
type ManagementRpcRequest,
type ManagementRpcResponse,
} from './management/types.ts';
import { resolveSlackPublicUrl } from './slack/credentials.ts';
import {
DurableRunDriver,
runDriverRetryDelayMs,
type RunDriverDrainResult,
} from './work/driver.ts';
import {
WorkStateError,
type WorkRpcRequest,
type WorkRpcResponse,
type WorkStore,
} from './work/types.ts';
import { runRoutineHeartbeat as runSharedRoutineHeartbeat } from './routines/heartbeat.ts';
// The generated default captures model and tool content. Register the native
// Cloudflare adapter explicitly for this Cloudflare-only entry so Workers
// Traces retain operational Flue spans without prompts, instructions, tool
// definitions, arguments, results, error messages, or stacks.
const cloudflareTracing = createCloudflareTracing({ content: false });
instrument({
...cloudflareTracing,
async observe(observation, context) {
emitRuntimeCorrelation(observation);
await cloudflareTracing.observe(observation, context);
emitManagementToolFailure(observation);
},
});
// This module is imported only by Flue's Cloudflare entry. Register before
// the generated entry's guarded default so `cloudflare/*` remains keyless but
// calls env.AI directly, without the default payload-logging AI Gateway.
// Importable `env` is Cloudflare's ambient binding object; no I/O runs here.
// A deployment serving many installations keeps the binding for its own
// judgments and registers no model provider from it.
registerCloudflareBindingProvider(env.AI, env as unknown as Record<string, unknown>);
export { ContainerProxy } from '@cloudflare/sandbox';
/**
* Cloudflare's Sandbox SDK routes intercepted container HTTPS through
* Worker-side handlers (sandbox/egress-outbound.ts). Profile grants are
* persisted as policy only; the credential is minted after each request
* passes the pure policy decision and is attached only to the Worker-side
* forwarded Request. On a deployment serving many installations the object
* is named under one (sandbox/sandbox-object.ts) and serves only it.
*/
export class Sandbox extends CloudflareSandbox<PlatformEnv> {
interceptHttps = true;
// Interception only covers HTTP and HTTPS. The container base class defaults
// `enableInternet` to true (@cloudflare/containers container.js:325), which
// leaves a raw, unmediated socket path alongside the intercepted one — a
// shell in this container could reach any host with `nc`, a raw TCP client,
// or DNS tunnelling and never touch `Sandbox.outbound`. The SDK's own
// outbound-interception example pairs `enableInternet = false` with
// `interceptHttps = true` for exactly this reason. The static catch-all
// below already forces intercept-all mode, so github.com, api.github.com and
// the allowlisted package registries keep flowing through the Worker
// handlers; this only removes the path that bypasses them.
enableInternet = false;
constructor(ctx: DurableObjectState, env: PlatformEnv) {
super(ctx, sandboxObjectEnv(ctx as unknown as SandboxObjectContext, env));
}
/**
* Reached only when the Containers SDK constructor succeeded, which requires
* a linked Container application. Starts nothing.
*/
async probeContainerRuntime(): Promise<boolean> {
return (this.ctx as { container?: unknown }).container !== undefined;
}
async prepareTurn(turnId: string): Promise<void> {
await this.policyState().prepareTurn(turnId);
}
async configureEgress(
input: SandboxEgressPolicyInput,
turnId: string,
): Promise<void> {
await this.policyState().configureEgress(input, turnId);
}
/**
* Decide whether this turn may reuse the running container. The workspace
* stays warm across turns for the same Agent and grants; any other owner
* gets a destroyed container rather than the prior checkout. Destroy only
* clears the SDK's own storage keys, so the prepared turn id survives.
*/
async beginWorkspaceTurn(input: {
fingerprint: string;
turnId: string;
}): Promise<{ state: WorkspaceTurnState; reservationId: string; restorable: boolean }> {
// A suspended or ended installation opens no workspace turn.
await requireWorkspaceTurnAdmitted(this.env);
await clearWarmWindow(this.warmWindowStorage());
const decision = await this.workspaceState().beginTurn({
...input,
containerRunning: this.containerRunning(),
now: Date.now(),
});
if (decision.retire) await this.destroy();
return {
state: decision.state,
reservationId: decision.reservationId,
restorable: decision.restorable && workspaceCheckpointsAvailable(this.env),
};
}
/**
* Admit this activation's container under the host's running limit and
* container-hours cap, holding its lease; standalone always admits.
* Called before the activation starts the container.
*/
async admitContainer(limits: SandboxContainerLimits): Promise<SandboxContainerAdmission> {
return admitWorkspaceContainer(this.env, {
key: this.ctx.id.toString(),
running: this.containerRunning(),
limits,
now: Date.now(),
});
}
/**
* Renew a running container's lease for its warm window, or release a
* stopped one's, metering its run: at a turn's end, and when the session
* cap refused a container admission had let start.
*/
async settleContainerLease(): Promise<void> {
await meterWorkspaceContainer(
this.env,
this.containerRunning() ? 'hold' : 'release',
{ key: this.ctx.id.toString(), now: Date.now() },
);
}
/** For work maintenance: whether a lease's container still runs. Starts nothing. */
async isContainerRunning(): Promise<boolean> {
return this.containerRunning();
}
/** The last start's lease hold, which a stop's release waits for. */
private startHold: Promise<void> | undefined;
override async onStart(): Promise<void> {
await super.onStart();
// The Containers SDK runs this inside blockConcurrencyWhile, so the hold
// and the admission check run beside the start, bounded, and every start
// is still metered.
const hold = holdStartedWorkspaceContainer(this.env, { key: this.ctx.id.toString(), now: Date.now() });
if (hold) {
this.startHold = hold;
this.waitUntil(hold);
}
// A session's next operation restarts a container the host stopped,
// through the SDK and past the activation's admission: a refused
// installation's start destroys itself.
const refusal = refuseUnadmittedContainerStart(this.env, () => this.destroy());
if (refusal) this.waitUntil(refusal);
}
/**
* The SDK's alarm stops an idle container at a deadline it keeps in
* memory, which every restart of this object pushes a full window out.
* The window a turn's end stored is the one that counts.
*/
override async alarm(alarmProps?: Parameters<CloudflareSandbox['alarm']>[0]): Promise<void> {
await stopPastWarmWindow({
storage: this.warmWindowStorage(),
running: this.containerRunning(),
now: Date.now(),
stop: () => this.onActivityExpired(),
});
await super.alarm(alarmProps);
}
override async onStop(params?: Parameters<CloudflareSandbox['onStop']>[0]): Promise<void> {
await super.onStop(params);
await this.releaseContainerLease();
}
/** Close a stopped container's lease, metering its run: its stop's, or an erasure's. */
private async releaseContainerLease(): Promise<void> {
// A start's hold still in flight lands first, so this release closes it.
await this.startHold;
await meterWorkspaceContainer(this.env, 'release', { key: this.ctx.id.toString(), now: Date.now() });
}
private waitUntil(promise: Promise<unknown>): void {
// The runtime has waitUntil on DurableObjectState; the module declaration omits it.
(this.ctx as DurableObjectState & { waitUntil(promise: Promise<unknown>): void }).waitUntil(promise);
}
/**
* Bring back the thread's last checkpoint into a cold container. Best
* effort: an expired, missing, or failed restore leaves an empty workspace
* and the Agent clones again.
*/
async restoreWorkspace(fingerprint: string): Promise<'restored' | 'unavailable'> {
return restoreWorkspaceCheckpoint({
env: this.env,
state: this.workspaceState(),
fingerprint,
now: Date.now,
restore: async (backup) => {
await this.restoreBackup(backup as Parameters<CloudflareSandbox['restoreBackup']>[0]);
},
});
}
/**
* Preset the Git author and committer for this workspace: the GitHub App's
* bot account, or the neutral Chickpea identity. Runs at each activation,
* because a cold container or a restored checkpoint has no global config.
* Identity only; credentials never enter Git configuration.
*/
async applyGitIdentity(): Promise<void> {
const identity = await resolveWorkspaceGitIdentity(getSettingsStore(this.env), fetch, this.env);
const result = await this.exec(gitIdentityConfigCommand(identity));
if (result.exitCode !== 0) throw new Error('Workspace Git identity was not applied');
}
/**
* End the turn's credential window without stopping the warm container,
* then checkpoint the workspace so the thread can resume after it sleeps.
*/
async endTurn(): Promise<void> {
await this.policyState().revokeEgress();
// Never start a container just to checkpoint it.
await checkpointWorkspace({
env: this.env,
containerRunning: this.containerRunning(),
state: this.workspaceState(),
now: Date.now,
create: () => this.createBackup({
dir: WORKSPACE_DIR,
ttl: WORKSPACE_CHECKPOINT_TTL_SECONDS,
excludes: [...WORKSPACE_CHECKPOINT_EXCLUDES],
localBucket: true,
}),
});
if (this.containerRunning()) {
const windowMs = await workspaceWarmWindowMs(this.env);
await holdWarmWindow(this.warmWindowStorage(), Date.now(), windowMs);
}
await this.settleContainerLease();
}
/**
* What the workspace tools report without a container round trip: whether
* the container is running and whether this owner has a checkpoint.
*/
async describeWorkspace(fingerprint: string): Promise<{ running: boolean; hasCheckpoint: boolean }> {
return {
running: this.containerRunning(),
hasCheckpoint:
workspaceCheckpointsAvailable(this.env) &&
(await this.workspaceState().hasCheckpoint(fingerprint, Date.now())),
};
}
/**
* Destroy the container and forget its checkpoint, deleting its objects,
* so the next turn starts from an empty workspace. The owner record and
* prepared turn survive.
*/
async discardWorkspace(): Promise<void> {
await discardWorkspaceCheckpoint({ env: this.env, state: this.workspaceState() });
await this.destroy();
}
/**
* The roster copy releases before thread continuity wrote here. Read only,
* to seed a thread instance once (see WorkspaceRosterSeed).
*/
async readWorkspaceRoster(key: string): Promise<WorkspaceRosterState> {
return readStoredWorkspaceRoster(this.policyStorage(), key);
}
/**
* A host turn's active coding-task records, kept on the thread's default
* workspace so a stop finds every coding job the run waits on.
*/
async readCodingTasks(turnId: string): Promise<CodingTaskRecordV1[]> {
return readStoredCodingTasks(this.policyStorage(), turnId);
}
async putCodingTask(turnId: string, record: unknown): Promise<void> {
await putStoredCodingTask(this.policyStorage(), turnId, record, Date.now());
}
async settleCodingTask(turnId: string, taskKey: string): Promise<void> {
await settleStoredCodingTask(this.policyStorage(), turnId, taskKey, Date.now());
}
// ── host functions (installation tenancy; see sandbox/sandbox-host.ts) ──
// An operator job's: each refuses on standalone and for any installation
// but the one this Sandbox's name scopes.
async chickpeaHostExportPage(request: ObjectExportRequest) {
return this.host().chickpeaHostExportPage(request);
}
async chickpeaHostErase(request: ObjectHostRequest) {
return this.host().chickpeaHostErase(request);
}
async chickpeaHostCancelPendingWork(request: ObjectHostRequest) {
return this.host().chickpeaHostCancelPendingWork(request);
}
async chickpeaHostStopContainer(request: ObjectHostRequest) {
return this.host().chickpeaHostStopContainer(request);
}
async chickpeaHostRestoreBookmarks(request: ObjectRestoreBookmarksRequest) {
return this.host().chickpeaHostRestoreBookmarks(request);
}
async chickpeaHostRestore(request: ObjectRestoreRequest) {
return this.host().chickpeaHostRestore(request);
}
async chickpeaHostRestoreRestart(request: ObjectRestoreRestartRequest) {
return this.host().chickpeaHostRestoreRestart(request);
}
private host(): SandboxHostRpc {
return sandboxHostFunctions({
env: this.env,
storage: this.ctx.storage as unknown as HostObjectStorage,
restoreContext: this.ctx,
running: () => this.containerRunning(),
destroy: () => this.destroy(),
stopRecorded: () => containerStopRecorded(() => this.getState()),
releaseLease: () => this.releaseContainerLease(),
currentCheckpoint: () => this.workspaceState().currentCheckpoint(),
containerState: () => this.getState(),
});
}
private containerRunning(): boolean {
return (this.ctx as { container?: { running?: boolean } }).container?.running === true;
}
private workspaceState(): SandboxWorkspaceState {
return new SandboxWorkspaceState(this.policyStorage());
}
private warmWindowStorage(): WarmWindowStorage {
return this.ctx.storage as unknown as WarmWindowStorage;
}
async getEgressPolicy(): Promise<SandboxEgressPolicy> {
return this.policyState().getEgressPolicy();
}
/**
* Everything an egress handler needs from this Sandbox in one call: the
* installation it serves (from its scoped env), its turn and the turn's
* policy.
*/
async egressContext(): Promise<SandboxEgressContext> {
const installationId = installationScopeOf(this.env)?.installationId;
return {
...(installationId ? { installationId } : {}),
...(await this.policyState().egressContext()),
};
}
async getTurnId(): Promise<string | undefined> {
return this.policyState().getTurnId();
}
async getTurnProgress(): Promise<TurnProgress> {
return this.policyState().getTurnProgress();
}
async recordPullRequestProgress(
pullRequest: TurnPullRequestProgress,
capturedTurnId: string,
): Promise<boolean> {
return this.policyState().recordPullRequestProgress(pullRequest, capturedTurnId);
}
private policyStorage(): SandboxPolicyStorage {
return this.ctx.storage as unknown as SandboxPolicyStorage;
}
private policyState(): SandboxPolicyState {
return new SandboxPolicyState(this.policyStorage());
}
}
// Assign through the SDK's inherited static setters so the handler registries
// are populated even when the Worker build preserves native class fields.
Sandbox.outboundByHost = {
'github.com': githubSandboxOutbound,
'api.github.com': githubSandboxOutbound,
...Object.fromEntries(
SANDBOX_PACKAGE_REGISTRY_HOSTS.map((host) => [host, packageRegistrySandboxOutbound]),
),
} satisfies Record<string, SandboxOutboundHandler>;
Sandbox.outbound = denySandboxOutbound;
// Backoff before the alarm re-fires for a job whose attempt failed but is not
// yet at the cap. A short delay (matching the DO alarm base retry) is enough:
// the failure that got here is a genuine delivery error, so an immediate retry
// would likely re-fail; a couple of seconds lets a transient Slack blip clear.
const RELAY_RETRY_BACKOFF_MS = 2_000;
const RELAY_BATCH_WINDOW_MS = 250;
/**
* Keep-warm: for KEEP_WARM_WINDOW_MS after a person's turn, the state store's
* alarm fires at least every KEEP_WARM_TICK_MS. A tick runs the ordinary idle
* drain (see keepWarmTickAt). The tick alone does not keep the store in
* memory: a store that left it made the next message pay 1.5-2.4 s at ingress
* for its first state read (production, 2026-11-10). Where the deployment
* sets a residency hold (ResidencyHold), the store holds itself for that
* window, and a store that restarted inside it holds itself again at its next
* tick (drainRelayAlarm). Constants, not settings; Node has no alarms.
*/
const KEEP_WARM_WINDOW_MS = 10 * 60_000;
const KEEP_WARM_TICK_MS = 20_000;
/**
* The management retention sweep (a 30-day cutoff) runs on an alarm firing at
* most this often per instance: the keep-warm ticks fire far more often than
* it can find anything. An instance that starts cold sweeps on its first firing.
*/
const RETENTION_SWEEP_INTERVAL_MS = 5 * 60_000;
// Pages of MAX_TURN_DRAIN_BATCH threads one alarm hands to thread runners.
// Each hand-off is one short admission RPC; the rest waits for the next alarm.
const RUNNER_DISPATCH_MAX_PAGES = 16;
/**
* Cloudflare entrypoint. Named exports of this file become top-level Worker
* exports on the CF target (the node target never imports it), so this is the
* ONE module allowed to import 'cloudflare:workers'.
*
* TagStateStore is the app-owned state Durable Object: a single named instance
* (state-rpc.ts TAG_STATE_INSTANCE) hosts all four store domains — config
* agents/assignments, thread snapshots, Slack claims + thread registry, and
* operator settings — by running the SAME target-neutral store logic classes
* the node backend runs, over DO SQLite instead of node:sqlite. Binding and
* migration live in wrangler.jsonc (TAG_STATE / migrations v2).
*/
type ReceiptOutboxWriter =
| Extract<ManagementRpcRequest['kind'],
'complete_setup' | 'put_outbox' | 'claim_introduction' | 'claim_agent_creation_welcome' | 'queue_owed_agent_welcome'>
| Extract<IdentityRpcRequest['kind'], 'claim_owner'>;
const RECEIPT_OUTBOX_WRITERS: ReadonlySet<string> = new Set<ReceiptOutboxWriter>([
'complete_setup', 'put_outbox', 'claim_introduction', 'claim_agent_creation_welcome', 'queue_owed_agent_welcome',
'claim_owner',
]);
export class TagStateStore extends DurableObject implements TagStateRpc, StateStoreHostRpc {
private stores: TagStateStores | undefined;
/**
* Constructor failures are latched instead of thrown: a throwing DO
* constructor makes EVERY subsequent RPC fail with an opaque platform 500.
* Latching turns that into a clear `{ok:false}` envelope per call that the
* proxies surface as a normal store error. The failure is NOT permanent for
* the isolate: `call()` re-attempts construction (a transient storage error
* on first boot should not brick every later RPC), so only the calls made
* before a successful re-init see the envelope.
*/
private initError: string | undefined;
/** Set while an alarm drains turns: new admissions start without waiting. */
private alarmAdmissionWake: (() => void) | undefined;
/**
* Turn jobs an alarm returned without at its hard cap (job id to thread
* key), still running in this isolate. Their threads stay closed until they
* settle so a delivery in flight is never started a second time.
*/
private readonly carriedAlarmTurns = new Map<string, string>();
/**
* Stop aborts this store sends for its own alarm's turns, per thread: the
* alarm starts no turn of a thread while one is out (see StopAbortFence).
* Replaced when the store is erased.
*/
private stopAborts = new StopAbortFence();
private readonly presentationRunnerOf = (runId: string) => this.presentationRunner(runId);
/** Set while runner-mode alarm work runs: admission hands new turns over at once. */
private dispatchWake: (() => void) | undefined;
/** Slack turns admitted in this isolate (events, deliveries, OAuth resumes). */
private admissionsSeen = 0;
/**
* When a person's turn was last admitted here (keepWarmTickAt). In memory
* only: an instance that starts cold reads it once from its newest turn
* row that is a person's (an Agent's ask or a setup wake is not one); 0
* when there is none.
*/
private lastHumanTurnAt: number | undefined;
/** Keeps this instance in memory after a person's turn (see hold()). */
private conversationHold: ResidencyHold | undefined;
/** When this instance last ran the management retention sweep (drainRelayAlarm). */
private lastRetentionSweepAt: number | undefined;
/**
* Turns whose first hand-off waits here for its status (showStatusAtHandoff).
* No re-admission overtakes one: it would hand the runner the presentation
* from before the status, then the first hand-off a later one.
*/
private readonly firstHandoffs = new Set<string>();
/** Direct deliveries this instance admits, one conversation at a time (admitSlackDelivery). */
private readonly directAdmissions = new ConversationOrder();
/**
* This instance, as the owner of the gateway inbox leases it takes. Only one
* instance owns this object's storage at a time; a lease under any other
* owner was taken by an instance that has been replaced (see
* GatewayInboxOwnership).
*/
private readonly instanceId = crypto.randomUUID();
/** This instance installed the schema: it is the first on this Worker version. */
private versionChanged = false;
/** The only caller of this installation's test run instances, loaded at the first test run. */
private testRunLifecycle: Promise<TestRunLifecycle> | undefined;
/** This activation's marks, logged as `store_boot` at its first RPC or alarm. */
private readonly boot = new StoreBootMarks();
constructor(ctx: DurableObjectState, env: unknown) {
super(ctx, objectInstallationEnv(ctx, env));
this.stores = this.tryInit();
// Work the previous instance was running stopped with it (a code update,
// a reset). Arm the alarm now rather than wait for the platform's retry.
void ctx.blockConcurrencyWhile(async () => {
await this.resumeAfterRestart();
this.boot.constructedOnVersion(this.versionChanged);
});
}
/**
* On a fresh instance, when nothing else will wake it soon: arm the alarm
* now if an alarm turn's Flue dispatch was in flight, a gateway delivery is
* leased by the previous instance, a test run's instance is not erased, or
* (first instance on a new Worker version) a hand-off to a thread runner
* is unconfirmed. Each signal is
* read on its own, so one failing read never hides the others. An alarm
* already set is kept: it is either due or a deliberate backoff (a rate
* limit's retry-after), which a restart must not skip. Never throws:
* blockConcurrencyWhile resets the object on a rejection.
*/
private async resumeAfterRestart(): Promise<void> {
const stores = this.stores;
if (!stores) return;
const read = (probe: () => boolean): boolean => {
try {
return probe();
} catch {
return false;
}
};
const alarmTurn = read(() => stores.turnJobs.hasInterruptedAlarmDispatch());
const inbox = read(() => stores.gatewayInbox.hasOrphanedLease());
const handoffs = this.versionChanged && read(() => stores.turnJobs.hasHandoffs());
const testRun = read(() => stores.testRuns.nextDueAt(Date.now()) !== undefined);
if (!alarmTurn && !inbox && !handoffs && !testRun) return;
try {
const armed = (await this.ctx.storage.getAlarm()) === null;
if (armed) await this.ctx.storage.setAlarm(Date.now());
console.info({
component: 'runtime',
event: 'state_store_resume',
versionChanged: this.versionChanged,
alarmTurn,
inbox,
handoffs,
testRun,
armed,
});
} catch {
console.warn('[chickpea] TagStateStore could not arm its resume alarm');
}
}
/** Execute one requester-bound management tool inside the state owner. The
* service and policy are identical to MCP/Admin/Node; only the Cloudflare
* transport changes so a compound Agent turn does not spend its Worker-loop
* budget on dozens of same-state proxy calls. */
async workspaceManagementInvoke(
request: SlackWorkspaceManagementRpcRequest,
): Promise<WorkspaceManagementToolResult> {
console.log('[chickpea:management] state RPC started', JSON.stringify({
tool: request.name,
}));
this.stores ??= this.tryInit();
const stores = this.stores;
if (!stores) {
console.error('[chickpea:management] state RPC unavailable', JSON.stringify({
tool: request.name,
}));
return workspaceManagementRpcFailure();
}
try {
const { local, service } = localManagementRuntime(
stores,
this.env as PlatformEnv,
);
const result = await invokeSlackWorkspaceManagementTool({
signal: request.signal,
identity: local.identity,
service,
name: request.name,
args: request.args,
});
try {
const outboxDueAt = stores.management.nextOutboxDueAt();
if (outboxDueAt !== undefined) {
await this.armAlarmNoLaterThan(Math.max(Date.now(), outboxDueAt));
}
} catch {
// The management operation has already reached a terminal service
// result. Outbox inspection and alarm delivery are retryable follow-up
// work and must not rewrite a successful mutation into an ambiguous
// transport error.
console.error('[chickpea] Workspace management receipt alarm failed');
}
console.log('[chickpea:management] state RPC completed', JSON.stringify({
tool: request.name,
outcome: result.ok ? 'success' : 'error',
...(!result.ok ? { reason: result.error.code } : {}),
}));
return result;
} catch (error) {
console.error('[chickpea] Workspace management state RPC failed', JSON.stringify({
tool: request.name,
errorName: error instanceof Error ? error.name : typeof error,
}));
return workspaceManagementRpcFailure();
}
}
/**
* Apply one approved management proposal for a Slack turn a thread runner
* executes. The runner holds no management state, so the approval crosses
* into the state owner once and runs against its local stores, exactly as
* the alarm executor's turns do. Non-idempotent: the caller never replays
* it, and the service's proposal state answers a re-run approval turn.
*/
async slackManagementApprovalInvoke(
request: SlackManagementApprovalRpcRequest,
): Promise<HostSlackManagementApprovalResult> {
this.stores ??= this.tryInit();
const stores = this.stores;
if (!stores) throw new Error('Management approval state is unavailable.');
const appStores = localGatewayAppStores(stores);
const { service } = localManagementRuntime(stores, this.env as PlatformEnv, appStores);
const result = await executeHostSlackManagementApproval({
turn: request.turn,
assignment: request.assignment,
turnJobId: request.turnJobId,
proposalId: request.proposalId,
dependencies: {
identity: appStores.identity,
config: appStores.config,
management: appStores.management,
service,
...(request.publicUrl ? { publicUrl: request.publicUrl } : {}),
// This store's own installation, never the runner's word for it.
...agentAvatarInstallationField(this.env as PlatformEnv),
},
...(request.presentationRunId ? { presentationRunId: request.presentationRunId } : {}),
});
try {
const outboxDueAt = stores.management.nextOutboxDueAt();
if (outboxDueAt !== undefined) {
await this.armAlarmNoLaterThan(Math.max(Date.now(), outboxDueAt));
}
} catch {
// The approval is applied; its receipt delivery is retryable follow-up.
console.error('[chickpea] Slack management approval receipt alarm failed');
}
return result;
}
async slackScheduleActionInvoke(
request: SlackScheduleActionRpcRequest,
): Promise<SlackScheduleActionOutcome> {
this.stores ??= this.tryInit();
const stores = this.stores;
if (!stores) throw new Error('Schedule action state is unavailable.');
const { local, service } = localManagementRuntime(
stores,
this.env as PlatformEnv,
);
const context = await resolveSlackManagementActor(request.signal, local.identity);
// Arm recovery before the first durable write so a DO interruption after
// admission cannot strand a pending action without an alarm.
await this.armAlarmNoLaterThan(Date.now());
const result = await scheduleActionRpcResult(() => invokeSlackScheduleAction({
signal: request.signal,
context,
operation: request.operation,
dependencies: {
management: local.management,
routines: local.routines,
service,
owner: `rpc:${request.signal.turnJobId}`,
},
}));
const nextAction = stores.routines.nextScheduleActionDueAt();
const nextReceipt = stores.management.nextOutboxDueAt();
const nextWake = earliestDefined(nextAction, nextReceipt);
if (nextWake !== undefined) await this.armAlarmNoLaterThan(Math.max(Date.now(), nextWake));
return result;
}
/**
* Build the store set over the DO's SQL storage, or latch the failure and
* return undefined. Idempotent by design so `call()` can re-run it to
* self-heal a failed first construction.
*/
private tryInit(): TagStateStores | undefined {
try {
// An object no installation's name addresses serves nothing.
requireInstallationScope(this.env as PlatformEnv);
const fingerprint = stateSchemaFingerprint(
cloudflareWorkerVersionId(this.env),
applicationIdentity,
{ localServe: viteServeLane },
);
const marker = fingerprint === undefined
? undefined
: new StateSchemaMarker(new DoSqlStateDb(this.ctx.storage), fingerprint);
if (marker?.isInstalled()) {
// This exact Worker version already completed and verified the
// install on this storage. Attach without DDL, migrations, probes or
// seeds: Durable Objects SQLite meters every row those statements
// read, and this constructor runs on every cold start.
this.initError = undefined;
return this.buildStores(new DoSqlStateDb(this.ctx.storage, 'attach'));
}
// The install and its marker commit atomically: constructor
// transactions (config seedOnce) nest as savepoints inside this one, and
// an uncaught throw anywhere discards every table, row and the marker,
// so the next cold start installs again. Any deploy, forward or
// rollback, carries a new upload id and installs once.
const db = new DoSqlStateDb(this.ctx.storage);
const stores = db.transaction(() => {
const built = this.buildStores(db);
marker?.record(Date.now());
return built;
});
if (marker) {
this.versionChanged = true;
console.info('[chickpea] TagStateStore schema installed', JSON.stringify({ fingerprint }));
}
this.initError = undefined;
return stores;
} catch (err) {
this.initError = err instanceof Error ? err.message : String(err);
console.error('[chickpea] TagStateStore init failed:', this.initError);
return undefined;
}
}
private buildStores(db: DoSqlStateDb): TagStateStores {
return buildTagStateStores(db, this.env as PlatformEnv, { gatewayLeaseOwner: this.instanceId });
}
/**
* Provisioning, on a deployment serving many installations: record the
* globally unique IDs its host assigned to this one. Idempotent; a store is
* never rebound to other IDs.
*/
async bindInstallation(
identity: InstallationIdentity,
): Promise<StateRpcResult<InstallationIdentity>> {
return this.call((stores) =>
bindStoreInstallation(stores.installationBinding, this.env as PlatformEnv, identity));
}
// ── host functions (installation tenancy; see state/installation-objects.ts) ──
// An operator job's: each refuses on standalone and for any installation
// but the one this store's name scopes (state/state-store-host.ts).
// Erasure must be the installation's last contact with this store: any
// later call, these included, builds an empty, seeded store again.
async chickpeaHostInventory(request: ObjectHostRequest & { cursor?: string | null; limit?: number }) {
return this.host().chickpeaHostInventory(request);
}
async chickpeaHostRestoreBookmarks(request: ObjectRestoreBookmarksRequest) {
return this.host().chickpeaHostRestoreBookmarks(request);
}
async chickpeaHostRestore(request: ObjectRestoreRequest) {
return this.host().chickpeaHostRestore(request);
}
async chickpeaHostRestoreRestart(request: ObjectRestoreRestartRequest) {
return this.host().chickpeaHostRestoreRestart(request);
}
async chickpeaHostInventoryBackfill(request: ObjectHostRequest) {
return this.host().chickpeaHostInventoryBackfill(request);
}
async chickpeaHostSandboxCensus(request: ObjectHostRequest & { now?: number }) {
return this.host().chickpeaHostSandboxCensus(request);
}
async chickpeaHostBrowserCensus(request: ObjectHostRequest & { now?: number }) {
return this.host().chickpeaHostBrowserCensus(request);
}
async chickpeaHostBrowserErase(request: ObjectHostRequest) {
return this.host().chickpeaHostBrowserErase(request);
}
async chickpeaHostExportPage(request: ObjectExportRequest) {
return this.host().chickpeaHostExportPage(request);
}
async chickpeaHostErase(request: ObjectHostRequest) {
return this.host().chickpeaHostErase(request);
}
async chickpeaHostCancelPendingWork(request: ObjectHostRequest) {
return this.host().chickpeaHostCancelPendingWork(request);
}
private host(): StateStoreHostRpc {
return stateStoreHostFunctions({
env: this.env as PlatformEnv,
storage: this.ctx.storage,
restoreContext: this.ctx,
stores: () => {
this.stores ??= this.tryInit();
if (!this.stores) throw new Error(`state store unavailable: init failed (${this.initError ?? 'unknown'})`);
return this.stores;
},
// Nothing of the erased store stays in memory: its stores, the alarm
// turns it carried and the stop aborts it fenced. A later call builds
// an empty store again.
onErased: () => {
this.stores = undefined;
this.carriedAlarmTurns.clear();
this.stopAborts = new StopAbortFence();
},
testRuns: () => this.testRuns(),
});
}
// ── config: agents ───────────────────────────────────────────────────────
async identityExecute(
request: IdentityRpcRequest,
): Promise<StateRpcResult<IdentityRpcResponse>> {
const result = this.call((stores) => stores.identity.execute(request));
if (result.ok) await this.armReceiptDrainAfter(request.kind);
return result;
}
async managementExecute(
request: ManagementRpcRequest,
): Promise<StateRpcResult<ManagementRpcResponse>> {
const result = this.call((stores) => stores.management.execute(request));
if (result.ok) await this.armReceiptDrainAfter(request.kind);
return result;
}
// Every request that can write a receipt outbox row arms the drain. The
// Agent welcome is claimed from inside the turn, which under the thread
// runner executes in its SlackThreadRunner, not in this object's alarm:
// nothing else would drain it until an unrelated wake.
private async armReceiptDrainAfter(kind: string): Promise<void> {
if (!RECEIPT_OUTBOX_WRITERS.has(kind)) return;
const due = this.call((stores) => stores.management.nextOutboxDueAt() ?? null);
if (due.ok && due.value !== null) {
await this.armAlarmNoLaterThan(Math.max(Date.now(), due.value));
}
}
async configListAgents(): Promise<StateRpcResult<CustomAgentConfig[]>> {
return this.call((stores) => stores.config.listAgents());
}
async configListUserAgents(): Promise<StateRpcResult<CustomAgentConfig[]>> {
return this.call((stores) => stores.config.listUserAgents());
}
async configListAgentAppBots(): Promise<StateRpcResult<AgentAppBot[]>> {
return this.call((stores) => stores.config.listAgentAppBots());
}
async configGetAgent(agentId: string): Promise<StateRpcResult<CustomAgentConfig>> {
return this.call((stores) => stores.config.getAgent(agentId));
}
async configGetAgentFirstAsk(agentId: string): Promise<StateRpcResult<string | null>> {
return this.call((stores) => stores.config.getAgentFirstAsk(agentId));
}
async configMaterializeChickpeaAgent(): Promise<StateRpcResult<CustomAgentConfig>> {
return this.call((stores) => stores.config.materializeChickpeaAgent());
}
async configCreateAgent(
agent: AgentCreateInput,
creationFence?: AgentCreationFence,
firstAsk?: string,
): Promise<StateRpcResult<CustomAgentConfig>> {
return this.call((stores) => stores.config.createAgent(agent, creationFence, firstAsk));
}
async configUpdateAgent(
agentId: string,
patch: ConfigAgentPatch,
expectedRevision?: number,
): Promise<StateRpcResult<CustomAgentConfig>> {
return this.call((stores) => stores.config.updateAgent(agentId, patch, expectedRevision));
}
async configMarkOAuthReauthorizationRequired(
target: OAuthReauthorizationTarget,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.markOAuthReauthorizationRequired(target));
}
async configMarkApiCredentialReady(target: ApiCredentialTarget): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.markApiCredentialReady(target));
}
async configDeleteAgent(
agentId: string,
expectedRevision?: number,
expectedCredentialIncarnation?: string,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.deleteAgent(
agentId,
expectedRevision,
expectedCredentialIncarnation,
));
}
async configDeleteAgentWithMemory(
agentId: string,
idempotencyKey: string,
expectedRevision: number,
expectedCredentialIncarnation: string,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.deleteAgentWithMemory(
agentId,
idempotencyKey,
expectedRevision,
expectedCredentialIncarnation,
stores.memory,
));
}
async configArchiveAgent(
agentId: string,
options?: {
replacementDefaultAgentId?: string;
expectedRevision?: number;
expectedCredentialIncarnation?: string;
},
): Promise<StateRpcResult<CustomAgentConfig>> {
return this.call((stores) => stores.config.archiveAgent(agentId, options));
}
async configRestoreAgent(
agentId: string,
expectedRevision?: number,
expectedCredentialIncarnation?: string,
): Promise<StateRpcResult<CustomAgentConfig>> {
return this.call((stores) => stores.config.restoreAgent(
agentId,
expectedRevision,
expectedCredentialIncarnation,
));
}
async configEnsureWorkspaceInstallation(
input: EnsureWorkspaceInstallationInput,
): Promise<StateRpcResult<WorkspaceInstallation>> {
return this.call((stores) => stores.config.ensureWorkspaceInstallation(input));
}
async configRetainGatewayInstallation(
input: import('./config/store.ts').RetainGatewayInstallationInput,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.retainGatewayInstallation(input));
}
async configRefreshGatewayClaimSetup(
input: import('./config/store.ts').RefreshGatewayClaimSetupInput,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.refreshGatewayClaimSetup(input));
}
async configGetWorkspaceInstallation(
workspaceId: string,
): Promise<StateRpcResult<WorkspaceInstallation | null>> {
return this.call((stores) => stores.config.getWorkspaceInstallation(workspaceId) ?? null);
}
async configListWorkspaceInstallations(): Promise<StateRpcResult<WorkspaceInstallation[]>> {
return this.call((stores) => stores.config.listWorkspaceInstallations());
}
async configUpdateWorkspaceInstallation(
workspaceId: string,
patch: WorkspaceInstallationPatch,
expectedRevision?: number,
): Promise<StateRpcResult<WorkspaceInstallation>> {
return this.call((stores) =>
stores.config.updateWorkspaceInstallation(workspaceId, patch, expectedRevision),
);
}
async configSetWorkspaceDefaultAgent(
workspaceId: string,
agentId: string,
expectedRevision?: number,
): Promise<StateRpcResult<WorkspaceInstallation>> {
return this.call((stores) =>
stores.config.setWorkspaceDefaultAgent(workspaceId, agentId, expectedRevision),
);
}
async configGetWorkspaceModelDefault(
workspaceId: string,
): Promise<StateRpcResult<WorkspaceModelDefault | null>> {
return this.call((stores) => stores.config.getWorkspaceModelDefault(workspaceId) ?? null);
}
async configPutWorkspaceModelDefault(
input: WorkspaceModelDefaultInput,
expectedRevision?: number,
): Promise<StateRpcResult<WorkspaceModelDefault>> {
return this.call((stores) => stores.config.putWorkspaceModelDefault(input, expectedRevision));
}
async configGetWorkspaceModelRole(
workspaceId: string,
role: NonChatModelRole,
): Promise<StateRpcResult<WorkspaceModelRole | null>> {
return this.call((stores) => stores.config.getWorkspaceModelRole(workspaceId, role) ?? null);
}
async configPutWorkspaceModelRole(
input: WorkspaceModelRoleInput,
expectedRevision?: number,
): Promise<StateRpcResult<WorkspaceModelRole>> {
return this.call((stores) => stores.config.putWorkspaceModelRole(input, expectedRevision));
}
async configGetAgentModelRole(
agentId: string,
role: NonChatModelRole,
): Promise<StateRpcResult<AgentModelRole | null>> {
return this.call((stores) => stores.config.getAgentModelRole(agentId, role) ?? null);
}
async configPutAgentModelRole(
input: AgentModelRoleInput,
expectedRevision?: number,
): Promise<StateRpcResult<AgentModelRole>> {
return this.call((stores) => stores.config.putAgentModelRole(input, expectedRevision));
}
async configUpdateAgentWithModelRoles(
agentId: string,
patch: ConfigAgentPatch,
roles: readonly AgentModelRolePatch[],
expectedRevision?: number,
): Promise<StateRpcResult<CustomAgentConfig>> {
return this.call((stores) =>
stores.config.updateAgentWithModelRoles(agentId, patch, roles, expectedRevision),
);
}
async configPrepareChickpeaCutover(
input: PrepareChickpeaCutoverInput,
): Promise<StateRpcResult<ChickpeaCutoverPreflight>> {
return this.call((stores) => stores.config.prepareChickpeaCutover(input));
}
async configPreflightChickpeaCutover(
workspaceId: string,
): Promise<StateRpcResult<ChickpeaCutoverPreflight>> {
return this.call((stores) => stores.config.preflightChickpeaCutover(workspaceId));
}
async configActivateChickpeaCutover(
input: ActivateChickpeaCutoverInput,
): Promise<StateRpcResult<ChickpeaCutoverActivation>> {
return this.call((stores) => stores.config.activateChickpeaCutover(input));
}
async configRollbackChickpeaCutover(
input: RollbackChickpeaCutoverInput,
): Promise<StateRpcResult<ChickpeaCutoverPreflight>> {
return this.call((stores) => stores.config.rollbackChickpeaCutover(input));
}
async configListAgentChannelGrants(
workspaceId?: string,
channelId?: string,
): Promise<StateRpcResult<AgentChannelGrant[]>> {
return this.call((stores) => stores.config.listAgentChannelGrants(workspaceId, channelId));
}
async configPutAgentChannelGrant(
input: AgentChannelGrantInput,
expectedRevision?: number,
): Promise<StateRpcResult<AgentChannelGrant>> {
return this.call((stores) => stores.config.putAgentChannelGrant(input, expectedRevision));
}
async configDeleteAgentChannelGrant(
workspaceId: string,
channelId: string,
agentId: string,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) =>
stores.config.deleteAgentChannelGrant(workspaceId, channelId, agentId),
);
}
async configGetAgentThreadRoute(
workspaceId: string,
channelId: string,
threadTs: string,
): Promise<StateRpcResult<AgentThreadRoute | null>> {
return this.call(
(stores) => stores.config.getAgentThreadRoute(workspaceId, channelId, threadTs) ?? null,
);
}
async configPutAgentThreadRoute(
input: AgentThreadRouteInput,
expectedRevision?: number,
): Promise<StateRpcResult<AgentThreadRoute>> {
return this.call((stores) => stores.config.putAgentThreadRoute(input, expectedRevision));
}
async configDeleteAgentThreadRoute(
workspaceId: string,
channelId: string,
threadTs: string,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) =>
stores.config.deleteAgentThreadRoute(workspaceId, channelId, threadTs)
);
}
async configListSlackPublicContext(
workspaceId: string,
channelId: string,
rootTs: string,
): Promise<StateRpcResult<SlackPublicContextEntry[]>> {
return this.call((stores) =>
stores.config.listSlackPublicContext(workspaceId, channelId, rootTs)
);
}
async configListRecentSlackPublicContext(
input: RecentSlackPublicContextInput,
): Promise<StateRpcResult<SlackPublicContextEntry[]>> {
return this.call((stores) => stores.config.listRecentSlackPublicContext(input));
}
async configListRecentSlackAgentReplies(
input: RecentSlackAgentRepliesInput,
): Promise<StateRpcResult<RecentSlackAgentReply[]>> {
return this.call((stores) => stores.config.listRecentSlackAgentReplies(input));
}
async configPutSlackPublicContext(
input: SlackPublicContextEntryInput,
): Promise<StateRpcResult<SlackPublicContextEntry>> {
return this.call((stores) => stores.config.putSlackPublicContext(input));
}
async configSeedSlackPublicContext(
inputs: SlackPublicContextEntryInput[],
): Promise<StateRpcResult<number>> {
return this.call((stores) => stores.config.seedSlackPublicContext(inputs));
}
async configDeleteSlackPublicContextMessage(
workspaceId: string,
channelId: string,
rootTs: string,
messageTs: string,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.deleteSlackPublicContextMessage(
workspaceId,
channelId,
rootTs,
messageTs,
));
}
async configDeleteSlackPublicContextRoot(
workspaceId: string,
channelId: string,
rootTs: string,
): Promise<StateRpcResult<number>> {
return this.call((stores) =>
stores.config.deleteSlackPublicContextRoot(workspaceId, channelId, rootTs)
);
}
async configListConnectionAccounts(
workspaceId: string,
): Promise<StateRpcResult<ConnectionAccount[]>> {
return this.call((stores) => stores.config.listConnectionAccounts(workspaceId));
}
async configPutConnectionAccount(
input: ConnectionAccountInput,
expectedRevision?: number,
): Promise<StateRpcResult<ConnectionAccount>> {
return this.call((stores) => stores.config.putConnectionAccount(input, expectedRevision));
}
async configCreateAgentOwnedConnection(
input: AgentOwnedConnectionInput,
): Promise<StateRpcResult<AgentOwnedConnection>> {
return this.call((stores) => stores.config.createAgentOwnedConnection(input));
}
async configApplySignInGroup(
change: SignInGroupChange,
): Promise<StateRpcResult<ConnectionAccount[]>> {
return this.call((stores) => stores.config.applySignInGroup(change));
}
async configReleaseSignInReference(input: {
connectionAccountId: string;
expectedRevision: number;
}): Promise<StateRpcResult<SignInReferenceRelease>> {
return this.call((stores) => stores.config.releaseSignInReference(input));
}
async configListAgentConnectionBindings(
agentId: string,
): Promise<StateRpcResult<AgentConnectionBinding[]>> {
return this.call((stores) => stores.config.listAgentConnectionBindings(agentId));
}
async configGetAgentConnectionBindingForAccount(
connectionAccountId: string,
): Promise<StateRpcResult<AgentConnectionBinding | null>> {
return this.call(
(stores) => stores.config.getAgentConnectionBindingForAccount(connectionAccountId) ?? null,
);
}
async configPutAgentConnectionBinding(
input: AgentConnectionBindingInput,
): Promise<StateRpcResult<AgentConnectionBinding>> {
return this.call((stores) => stores.config.putAgentConnectionBinding(input));
}
async configPutAgentConnectionBindingIfUnchanged(
expected: AgentConnectionBinding,
next: AgentConnectionBindingInput,
): Promise<StateRpcResult<AgentConnectionBinding | null>> {
return this.call((stores) => stores.config.putAgentConnectionBindingIfUnchanged(expected, next) ?? null);
}
async configListAgentScheduleReferences(
agentId: string,
): Promise<StateRpcResult<AgentScheduleReference[]>> {
return this.call((stores) => stores.config.listAgentScheduleReferences(agentId));
}
async configSummarizeAdoptionInventory() {
return this.call((stores) => stores.config.summarizeAdoptionInventory());
}
async configGetAgentScheduleReference(
scheduleId: string,
): Promise<StateRpcResult<AgentScheduleReference | null>> {
return this.call((stores) => stores.config.getAgentScheduleReference(scheduleId) ?? null);
}
async configPutAgentScheduleReference(
input: AgentScheduleReferenceInput,
expectedRevision?: number,
): Promise<StateRpcResult<AgentScheduleReference>> {
return this.call((stores) => stores.config.putAgentScheduleReference(input, expectedRevision));
}
async configRetireAgentScheduleReference(
scheduleId: string,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.retireAgentScheduleReference(scheduleId));
}
async configListChannels(): Promise<StateRpcResult<ChannelConfig[]>> {
return this.call((stores) => stores.config.listChannels());
}
async configGetChannel(
workspaceId: string,
channelId: string,
): Promise<StateRpcResult<ChannelConfig | null>> {
return this.call((stores) => stores.config.getChannel(workspaceId, channelId) ?? null);
}
async configPutChannel(
channel: ChannelConfig,
expectedRevision?: number,
): Promise<StateRpcResult<ChannelConfig>> {
return this.call((stores) => stores.config.putChannel(channel, expectedRevision));
}
async configRefreshChannelLabel(
workspaceId: string,
channelId: string,
label: string,
): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.config.refreshChannelLabel(workspaceId, channelId, label));
}
// ── config: assignments ──────────────────────────────────────────────────
async configGetAgentReferences(
agentId: string,
): Promise<StateRpcResult<AgentReferenceSummary>> {
return this.call((stores) => stores.config.getAgentReferences(agentId));
}
// ── config: Agent thread snapshots ──────────────────────────────────────
async snapshotGet(threadKey: string): Promise<StateRpcResult<AgentSnapshot | null>> {
return this.call((stores) => stores.snapshots.get(threadKey) ?? null);
}
async snapshotPutIfAbsent(
threadKey: string,
snapshot: AgentSnapshot,
): Promise<StateRpcResult<AgentSnapshot>> {
return this.call((stores) => stores.snapshots.putIfAbsent(threadKey, snapshot));
}
async snapshotReplace(
threadKey: string,
snapshot: AgentSnapshot,
): Promise<StateRpcResult<AgentSnapshot>> {
return this.call((stores) => stores.snapshots.replace(threadKey, snapshot));
}
async snapshotListLiveRootsByAgent(
agentId: string,
): Promise<StateRpcResult<AgentSnapshotRootReference[]>> {
return this.call((stores) => stores.snapshots.listLiveRootsByAgent(agentId));
}
// ── slack claims + thread registry ───────────────────────────────────────
async claim(key: string): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.slack.claim(key));
}
async release(key: string): Promise<StateRpcResult<null>> {
return this.call((stores) => {
stores.slack.release(key);
return null;
});
}
async threadStart(key: string): Promise<StateRpcResult<null>> {
return this.call((stores) => {
stores.slack.start(key);
return null;
});
}
async threadHas(key: string): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.slack.has(key));
}
async threadActiveWorkGet(key: string): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.slack.isActiveWork(key));
}
async threadActiveWorkSet(
key: string,
generation: string,
active: boolean,
): Promise<StateRpcResult<null>> {
return this.call((stores) => {
stores.slack.setActiveWork(key, generation, active);
return null;
});
}
async admitSlackTurn(input: SlackCanonicalAdmissionInput) {
const result = this.call((stores) =>
stores.slack.admitCanonical(
input, stores.work, stores.turnJobs, stores.presentations, stores.uiSurfaces,
),
);
if (result.ok && result.value.claimed && 'steered' in result.value) {
await this.offerCreatedStop(result.value.steered);
}
if (result.ok && input.turnJob && 'admission' in result.value) await this.afterSlackAdmission(input.turnJob);
return result;
}
/** A just-admitted turn: hold this store, hand the turn over with its first status, arm the alarm. */
private async afterSlackAdmission(turnJob: TurnJob): Promise<void> {
// A person's turn opens the keep-warm window; the host's own turns, an
// Agent's ask and a setup wake, do not.
const { agentAsk, setupWake } = turnJob.turn;
if (!agentAsk && !setupWake) {
this.lastHumanTurnAt = Date.now();
this.hold().holdFrom(this.lastHumanTurnAt);
}
const ready = this.runnerMode()
? this.call((stores) => stores.turnJobs.dispatchableNow(turnJob.id))
: undefined;
const handedOver = ready?.ok && ready.value
? await this.admitToRunner(ready.value, true).catch(() => false)
: false;
if (!handedOver) {
await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
} else {
// The runner has the turn; the alarm only ticks for the keep-warm
// window, so a running alarm's dispatch loop is not woken for it.
const tick = this.keepWarmTickAt();
if (tick !== undefined) await this.setAlarmNoLaterThan(tick);
}
}
async slackTurnAdmitEdit(edit: SlackMessageEdit) {
return this.call((stores) =>
stores.slack.admitEdit(edit, stores.work, stores.turnJobs, stores.presentations));
}
async uiSurfaceExecute(request: UiSurfaceRpcRequest) {
return this.call((stores) => stores.uiSurfaces.execute(request));
}
async slackAgentBindingPin(
input: Parameters<TagStateRpc['slackAgentBindingPin']>[0],
expected?: Parameters<TagStateRpc['slackAgentBindingPin']>[1],
) {
return this.call((stores) => stores.turnJobs.pinAgentBinding(input, expected));
}
async slackAgentBindingGet(continuityKey: string) {
return this.call((stores) =>
stores.turnJobs.getAgentBinding(continuityKey) ?? null,
);
}
async slackFlueDispatchPrepare(
id: string,
message: string,
observation: Parameters<TagStateRpc['slackFlueDispatchPrepare']>[2],
threadImages?: Parameters<TagStateRpc['slackFlueDispatchPrepare']>[3],
admittedListIds?: Parameters<TagStateRpc['slackFlueDispatchPrepare']>[4],
turnEnvelope?: Parameters<TagStateRpc['slackFlueDispatchPrepare']>[5],
threadFileIds?: Parameters<TagStateRpc['slackFlueDispatchPrepare']>[6],
workspaceChangeTools?: Parameters<TagStateRpc['slackFlueDispatchPrepare']>[7],
) {
return this.call((stores) =>
stores.turnJobs.prepareFlueDispatch(
id, message, observation, threadImages, admittedListIds, turnEnvelope, threadFileIds, workspaceChangeTools,
),
);
}
async slackTurnEnvelopeGet(id: string) {
return this.call((stores) => stores.turnJobs.getTurnEnvelope(id) ?? null);
}
async slackWorkspaceObjectRecord(object: InstallationWorkspaceObject) {
return this.call((stores) => {
stores.objectInventory.recordWorkspaceObject(object);
return null;
});
}
async slackFlueExistingInstanceReconcile(id: string, uid: string) {
return this.call((stores) => stores.turnJobs.reconcileFlueExistingInstance(id, uid));
}
async slackFlueReceiptRecord(
id: string,
receipt: Parameters<TagStateRpc['slackFlueReceiptRecord']>[1],
) {
return this.call((stores) => stores.turnJobs.recordFlueReceipt(id, receipt));
}
async slackFlueSettlementRecord(
id: string,
settlement: Parameters<TagStateRpc['slackFlueSettlementRecord']>[1],
) {
return this.call((stores) => stores.turnJobs.recordFlueSettlement(id, settlement));
}
async slackFlueObservationMatch(instanceId: string, submissionId?: string) {
return this.call((stores) =>
stores.turnJobs.matchFlueObservation(instanceId, submissionId) ?? null,
);
}
/** The model-access lookup's binding of a test run's attempt to its live row. */
async testRunModelAccessBind(
instanceId: string,
submissionId: string,
): Promise<StateRpcResult<{ runtimePlan: unknown } | null>> {
const bound = this.call((stores) => stores.testRuns.bindModelAccess(instanceId, submissionId));
if (!bound.ok) return bound;
// A submission that landed after its run's erase unerased the run: the alarm erases it again.
if (bound.value?.reopened) await this.armAlarmNoLaterThan(Date.now());
return { ok: true, value: bound.value && 'runtimePlan' in bound.value ? { runtimePlan: bound.value.runtimePlan } : null };
}
async testRunStart(request: TestRunStartRequest): Promise<StateRpcResult<TestRunStartResult>> {
return this.callTestRuns(async () => (await this.testRuns()).start(request));
}
async testRunReplay(query: TestRunReplayQuery): Promise<StateRpcResult<TestRunView | null>> {
return this.callTestRuns(async () => await (await this.testRuns()).replay(query) ?? null);
}
async testRunObserve(query: TestRunQuery): Promise<StateRpcResult<TestRunView | null>> {
return this.callTestRuns(async () => await (await this.testRuns()).observe(query) ?? null);
}
async testRunList(query: PracticeRunsQuery): Promise<StateRpcResult<PracticeRunListing[]>> {
return this.callTestRuns(async () => (await this.testRuns()).list(query));
}
/** One lifecycle per object: it holds which runs' steps are going and which it aborted. */
private testRuns(): Promise<TestRunLifecycle> {
this.testRunLifecycle ??= import('./state/test-run-lifecycle.ts').then(({ TestRunLifecycle }) => new TestRunLifecycle({
stores: () => {
this.stores ??= this.tryInit();
if (!this.stores) throw new Error(`state store unavailable: init failed (${this.initError ?? 'unknown'})`);
return this.stores;
},
port: async () => {
const { cloudflareTestRunInstancePort } = await import('./state/test-run-instance.ts');
return cloudflareTestRunInstancePort(this.env as PlatformEnv);
},
armAlarm: (at) => this.armAlarmNoLaterThan(at),
platformEnv: this.env as PlatformEnv,
}));
return this.testRunLifecycle;
}
private async callTestRuns<T>(work: () => Promise<T>): Promise<StateRpcResult<T>> {
try {
return { ok: true, value: await work() };
} catch (error) {
console.error('[chickpea] test run state RPC failed', JSON.stringify({
errorName: error instanceof Error ? error.name : typeof error,
}));
return { ok: false, error: { code: 'internal', message: 'The test run could not be reached.' } };
}
}
async slackTurnRecoveryRequired(
id: string,
reason: string,
): Promise<StateRpcResult<null>> {
return this.call((stores) => {
stores.turnJobs.markRecoveryRequired(id, reason);
return null;
});
}
async slackTurnRecoveryList(limit: number) {
return this.call((stores) => stores.turnJobs.listRecoveryRequired(limit));
}
async slackInstallationRecoveryRetry(workspaceId: string) {
const result = this.call((stores) =>
stores.turnJobs.retrySlackInstallationRecovery(workspaceId),
);
if (result.ok && result.value > 0) {
await this.armAlarmNoLaterThan(Date.now() + RELAY_BATCH_WINDOW_MS);
}
return result;
}
async slackTurnRecoveryResolve(id: string) {
return this.call((stores) => stores.turnJobs.resolveRecoveryRequired(id));
}
async slackInstallationPendingDeliveryCount(workspaceId: string) {
return this.call((stores) =>
stores.turnJobs.countPendingDeliveriesForWorkspace(workspaceId),
);
}
async slackInteractionProgressRecord(
id: string,
patch: Parameters<TagStateRpc['slackInteractionProgressRecord']>[1],
): Promise<StateRpcResult<null>> {
return this.call((stores) => {
stores.turnJobs.recordSlackInteractionProgress(id, patch);
return null;
});
}
async slackPresentationGet(runId: string) {
const runner = this.presentationRunner(runId);
if (runner) return runner.presentationGet(runId);
return this.call((stores) => stores.presentations.get(runId) ?? null);
}
async slackProposalApprovalTurns(input: Parameters<TagStateRpc['slackProposalApprovalTurns']>[0]) {
return this.call((stores) => stores.turnJobs.listProposalApprovalTurns(input));
}
async slackLatestAgentTurn(input: Parameters<TagStateRpc['slackLatestAgentTurn']>[0]) {
return this.call((stores) => stores.turnJobs.latestAgentTurnMessageTs(input) ?? null);
}
async slackTypedMessage(input: Parameters<TagStateRpc['slackTypedMessage']>[0]) {
return this.call((stores) => stores.turnJobs.typedSlackMessage(input) ?? null);
}
async slackPresentationLatestThreadGeneration(
root: Parameters<TagStateRpc['slackPresentationLatestThreadGeneration']>[0],
) {
return this.call((stores) =>
stores.presentations.getLatestThreadSessionGeneration(root) ?? null,
);
}
async slackPresentationTransition(
input: Parameters<TagStateRpc['slackPresentationTransition']>[0],
) {
const runner = this.presentationRunner(input.runId);
if (runner) return runner.presentationTransition(input);
return this.call((stores) => stores.presentations.transition(input));
}
/**
* The thread runner holding the authoritative copy of a run's presentation,
* when a runner executes its turn. Presentation effects from outside the
* turn (an Agent welcome settling a deferred terminal) go there, so the
* runner and this store never advance two copies.
*/
private presentationRunner(runId: string): SlackThreadRunnerRpc | undefined {
try {
this.stores ??= this.tryInit();
const presentation = this.stores?.presentations.get(runId);
const threadKey = presentation && this.stores!.turnJobs.runnerThreadKey(
presentation.turnJobId,
slackAgentThreadKey,
);
return threadKey ? threadRunnerStub(this.env as PlatformEnv, threadKey) : undefined;
} catch {
return undefined;
}
}
async slackPresentationReserveAppend(workspaceId: string) {
return this.call((stores) => stores.presentations.reserveAppend(workspaceId));
}
async slackPresentationAppendCooldown(workspaceId: string) {
return this.call((stores) => stores.presentations.appendCooldownUntil(workspaceId) ?? null);
}
async slackPresentationApplyCooldown(workspaceId: string, retryAfterMs: number) {
return this.call((stores) =>
stores.presentations.applyAppendCooldown(workspaceId, retryAfterMs),
);
}
async slackPresentationReserveActivityStatus(workspaceId: string) {
return this.call((stores) => stores.presentations.reserveActivityStatus(workspaceId));
}
async slackPresentationApplyActivityStatusCooldown(
workspaceId: string,
retryAfterMs: number,
) {
return this.call((stores) =>
stores.presentations.applyActivityStatusCooldown(workspaceId, retryAfterMs),
);
}
async slackReserveRead(workspaceId: string, method: SlackReadMethod) {
return this.call((stores) => stores.presentations.reserveSlackRead(workspaceId, method));
}
async slackApplyReadCooldown(
workspaceId: string,
method: SlackReadMethod,
retryAfterMs: number,
) {
return this.call((stores) =>
stores.presentations.applySlackReadCooldown(workspaceId, method, retryAfterMs),
);
}
async slackPresentationRepairList(limit: number) {
return this.call((stores) => stores.presentations.listAutoRepairableV3(limit));
}
async slackPresentationMaintain(limit: number) {
return this.call((stores) => stores.presentations.maintain(limit));
}
async slackPresentationSummary(workspaceId: string) {
return this.call((stores) => stores.presentations.summarize(workspaceId));
}
// ── operator settings ────────────────────────────────────────────────────
async settingGet(key: string): Promise<StateRpcResult<string | null>> {
return this.call((stores) => stores.settings.getSetting(key) ?? null);
}
async settingGetMany(keys: readonly string[]): Promise<StateRpcResult<(string | null)[]>> {
return this.call((stores) => stores.settings.getSettings(keys).map((value) => value ?? null));
}
async settingSet(key: string, value: string): Promise<StateRpcResult<null>> {
return this.call((stores) => {
stores.settings.setSetting(key, value);
return null;
});
}
async settingDelete(key: string): Promise<StateRpcResult<null>> {
return this.call((stores) => {
stores.settings.deleteSetting(key);
return null;
});
}
async settingApplyPatch(patch: SettingsPatch): Promise<StateRpcResult<boolean>> {
return this.call((stores) => stores.settings.applySettingsPatch(patch));
}
async settingMergeStringSet(
key: string,
values: readonly string[],
): Promise<StateRpcResult<string[]>> {
return this.call((stores) => stores.settings.mergeSettingStringSet(key, values));
}
async encryptedCredentialGet(key: string) {
return this.call((stores) => stores.settings.getEncryptedCredentialRevision(key) ?? null);
}
async encryptedCredentialReplace(input: ReplaceEncryptedCredentialRevisionInput) {
return this.call((stores) => stores.settings.replaceEncryptedCredentialRevision(input) ?? null);
}
async encryptedCredentialCensus() {
return this.call((stores) => stores.settings.censusEncryptedCredentialRevisions());
}
async encryptedCredentialList(prefix: string) {
return this.call((stores) => stores.settings.listEncryptedCredentialRevisions(prefix));
}
async encryptedCredentialDelete(key: string, expectedRevision: string) {
return this.call((stores) => stores.settings.deleteEncryptedCredentialRevision(
key,
expectedRevision,
));
}
async modelCredentialRead(providerId: string) {
return this.call((stores) => stores.settings.readModelCredential(providerId) ?? null);
}
async modelCredentialPublish(input: PublishModelCredentialInput) {
return this.call((stores) => stores.settings.publishModelCredential(input));
}
async modelCredentialRewrap(input: RewrapModelCredentialInput) {
return this.call((stores) => stores.settings.rewrapModelCredential(input));
}
// ── memory + generic audit envelope ─────────────────────────────────────
async memoryExecute(
request: MemoryRpcRequest,
): Promise<StateRpcResult<MemoryRpcResponse>> {
return this.call((stores) => stores.memory.execute(request));
}
async routinesExecute(
request: RoutineRpcRequest,
): Promise<StateRpcResult<RoutineRpcResponse>> {
return this.call((stores) => stores.routines.execute(request));
}
async usageExecute(
request: UsageRpcRequest,
): Promise<StateRpcResult<UsageRpcResponse>> {
return this.call((stores) => stores.usage.execute(request));
}
async workExecute(
request: WorkRpcRequest,
): Promise<StateRpcResult<WorkRpcResponse>> {
return this.call((stores) => stores.work.execute(request));
}
async runtimeDrainStatus(): Promise<StateRpcResult<RuntimeDrainStatus>> {
return this.call((stores) => {
const categories = {
...stores.turnJobs.runtimeDrainCounts(),
...stores.gatewayInbox.runtimeDrainCounts(),
executingRuns: stores.work.countExecutingRuns(),
admittingOrRunningRoutineOccurrences:
stores.routines.countAdmittingOrRunningOccurrences(),
};
return buildRuntimeDrainStatus(categories);
});
}
async maintainWork(at: number): Promise<StateRpcResult<null>> {
if (!Number.isSafeInteger(at) || at < 0) {
return rpcError('work', 'Work maintenance time is invalid.', {
workCode: 'work_maintenance_invalid',
});
}
const result = this.call((stores) => {
stores.work.purgeContent(at, 100);
stores.presentations.maintain(100);
return stores.turnJobs.hasPending('legacy') || stores.turnJobs.hasPending('ledger');
});
if (!result.ok) return result;
// Maintenance repairs a missing wake; it does not admit new work. Keep an
// existing rate-limit or delivery backoff instead of restarting it early.
if (result.value && (await this.ctx.storage.getAlarm()) === null) {
await this.ctx.storage.setAlarm(Date.now() + RELAY_BATCH_WINDOW_MS);
}
return { ok: true, value: null };
}
// ── turn relay (Cloudflare turn-horizon fix) ─────────────────────────────
async enqueueTurn(job: TurnJob): Promise<StateRpcResult<null>> {
const result = this.call((stores) => {
stores.turnJobs.enqueue(job);
return null;
});
// Arm the alarm only after the row is written, and AWAIT it: the job + the
// armed alarm must both be durable before this RPC resolves, because the
// events handler acks Slack the instant it does. Under the alarm executor
// a small, non-sliding batch window lets near-simultaneous independent
// threads reach the existing bounded fan-out. Bring a later receipt/retry
// alarm forward for new work, but never move an already-armed alarm later.
if (result.ok) {
await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
}
return result;
}
async slackTurnSteer(
request: Parameters<TagStateRpc['slackTurnSteer']>[0],
enqueue?: Parameters<TagStateRpc['slackTurnSteer']>[1],
): ReturnType<TagStateRpc['slackTurnSteer']> {
const result = this.call((stores) => stores.turnJobs.steer(request, enqueue));
if (!result.ok) return result;
const decision = result.value;
if (decision.outcome === 'enqueue' && decision.enqueued) {
// As enqueueTurn: the row and its alarm are durable before the reply.
await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
} else {
await this.offerCreatedStop(decision);
}
return result;
}
/**
* A stop the RPC just created (a steer, or a typed stop the admission
* recorded) is offered to its runner before the RPC answers; the outbox
* retries the rest. Any other decision, and a stop that already existed,
* offers nothing.
*/
private async offerCreatedStop(decision: TurnSteeringDecision): Promise<void> {
if (decision.outcome === 'stopped' && decision.stop.created) await this.deliverStopNotices();
}
async slackTurnStopFinish(
headId: string,
outcome: 'dropped' | 'released',
): ReturnType<TagStateRpc['slackTurnStopFinish']> {
return this.call((stores) => stores.slack.finishTurnStop(
headId,
outcome,
stores.turnJobs,
stores.work,
stores.presentations,
) ?? null);
}
/**
* The alarm executor runs rows it listed before the turns ahead of them
* ran, so a follow-up its head took could run again: only a head its
* thread runner runs, which reads each row again first, takes any.
*/
async slackTurnFold(request: TurnFoldRequest): ReturnType<TagStateRpc['slackTurnFold']> {
return this.call((stores) => stores.turnJobs.runnerView(request.headId).job?.executor === 'runner'
? stores.turnJobs.fold(request)
: { messages: [], requesterText: '' });
}
async slackTurnReadThrough(headId: string): ReturnType<TagStateRpc['slackTurnReadThrough']> {
return this.call((stores) => stores.turnJobs.readThrough(headId) ?? null);
}
async slackTurnDirectThreads(
query: Parameters<TagStateRpc['slackTurnDirectThreads']>[0],
): ReturnType<TagStateRpc['slackTurnDirectThreads']> {
return this.call((stores) => stores.turnJobs.runningDirectThreadKeys(query));
}
/** The alarm executor registers its turns in this isolate's status registry. */
async slackRunFacts(turnJobId: string): ReturnType<TagStateRpc['slackRunFacts']> {
return this.call(() => {
if (typeof turnJobId !== 'string' || turnJobId.length === 0 || turnJobId.length > 256) {
throw new Error('TurnJob id is invalid.');
}
return defaultSlackStatusRegistry.runFactsView(turnJobId) ?? null;
});
}
/**
* The stop outbox (KTD2): offer every due stop to its head row's thread
* runner, whose acknowledgement clears it, like a runner hand-off admitted
* again until confirmed. A head no runner owns yet stays owed: the alarm
* hands it over first, and its dispatch is refused meanwhile. A head this
* store's own alarm executes (SLACK_TAG_TURN_EXECUTOR=alarm) is aborted
* here instead (receiveAlarmExecutorStop), inside its thread's abort fence.
* Every offer is bounded, so a wedged runner or coordinator never holds the
* steer and admission RPCs or the relay alarm that await this: one not
* answered in time counts as not acknowledged and is retried. Keeps the
* alarm armed for notices still owed. Never throws.
*/
private async deliverStopNotices(): Promise<void> {
const stores = this.stores;
if (!stores) return;
try {
await deliverDueStopNotices({
turnJobs: stores.turnJobs,
receiver: async (notice) => {
if (notice.executor !== 'runner') {
return this.stopAborts.run(notice.runnerKey, () => receiveAlarmExecutorStop(
stores.turnJobs.runnerView(notice.turnJobId),
(target) => abortSlackThreadAgent(target),
() => stores.turnJobs.runnerView(notice.turnJobId),
));
}
// A runner of an older version has no `stop`: the call rejects
// and the notice stays owed until its successor takes it.
const runner = threadRunnerStub(this.env as PlatformEnv, notice.runnerKey);
if (!runner) return false;
const taken = await boundedStopCall(runner.stop(notice), STOP_NOTICE_DELIVERY_TIMEOUT_MS);
return taken.acknowledged === true;
},
});
const due = stores.turnJobs.nextStopNoticeDueAt();
if (due !== undefined) await this.armAlarmNoLaterThan(Math.max(Date.now(), due));
} catch {
console.warn('[chickpea] stop notice delivery failed; the alarm retries it');
}
}
async receiveGatewayHttp(input: {body: string; signature: string; url: string}): Promise<{status: number; body: unknown}> {
const receivedAt = Date.now();
let observed: {delivery: GatewayInboundDelivery; issuedAt: number} | undefined;
let outcome: 'accepted' | 'duplicate' | 'failed' = 'failed';
try {
const snapshot = this.call(stores => ({
binding: stores.settings.getSetting(GATEWAY_BINDING_SETTING),
delivery: stores.settings.getSetting(GATEWAY_HTTP_SETTING),
}));
if (!snapshot.ok || !snapshot.value.binding || !snapshot.value.delivery) throw new HttpDeliveryError(503, 'delivery_not_configured');
const binding = JSON.parse(snapshot.value.binding) as GatewayWorkspaceBinding;
const state = parseHttpDeliveryState(snapshot.value.delivery)!;
const value = await verifyHttpDelivery({...input, binding, state, keyring:loadCredentialKeyring(this.env as PlatformEnv)});
if (value.kind === 'gateway.delivery') observed = {delivery:value.delivery!, issuedAt:value.issuedAt};
const keyForm = value.kind === 'gateway.delivery' ? await loadKeyFormFor(value.delivery!) : undefined;
// Crypto yields. Recheck the exact settings and installation in the owning
// state turn, with no await between authorization and insertion.
let admissionError: unknown;
const admitted = this.call(stores => {
try {
if (stores.settings.getSetting(GATEWAY_BINDING_SETTING) !== snapshot.value.binding ||
stores.settings.getSetting(GATEWAY_HTTP_SETTING) !== snapshot.value.delivery) throw new HttpDeliveryError(409, 'delivery_state_changed');
const installation = stores.config.getWorkspaceInstallation(binding.workspaceId);
if (stores.identity.getAuthControl()?.healthGate === 'recovery_only' || !installation ||
installation.transportMode !== 'gateway' || installation.health === 'revoked' ||
installation.gatewayBindingId !== binding.bindingId || installation.appId !== binding.appId ||
installation.botUserId !== binding.botUserId) throw new HttpDeliveryError(409, 'delivery_binding_rejected');
if (value.kind === 'gateway.challenge') return 'verified' as const;
// A modal click or submission is answered in the receipt from its stored card.
const receipt = keyForm
? keyForm.keyFormReceiptNow(value.delivery!, stores)
: gatewayUiReceiptNow(value.delivery!, stores.uiSurfaces);
const outcome = receipt ?? stores.gatewayInbox.admit(value.delivery!);
// Only a real delivery proves gateway activation. A challenge must not
// retire the old route before the gateway commits its compare-and-swap.
if (state.pending?.keyId === value.keyId && state.pending.routeRevision === value.routeRevision) {
stores.settings.setSetting(GATEWAY_HTTP_SETTING, JSON.stringify({version:1,bindingId:binding.bindingId,deploymentId:binding.deploymentId,installedAt:binding.installedAt,
mode:'http',revision:value.routeRevision,active:state.pending}));
}
return outcome;
} catch (error) { admissionError = error; throw error; }
});
if (!admitted.ok) {
if (admissionError instanceof HttpDeliveryError) throw admissionError;
if (admissionError instanceof GatewayInboxConflictError) throw new HttpDeliveryError(409,'delivery_identity_conflict');
throw new HttpDeliveryError(503,'delivery_unavailable');
}
if (typeof admitted.value !== 'string') {
outcome = admitted.value.outcome;
return {status:200, body:{...httpDeliveryReceipt(value, admitted.value.outcome), interaction:admitted.value.interaction}};
}
if (admitted.value !== 'verified') outcome = admitted.value;
if (admitted.value === 'accepted') this.wakeRunnerAtReceipt(value.delivery!);
// A duplicate also arms recovery: a previous insert may have survived an
// alarm-write failure or loss of the HTTP response.
if (admitted.value !== 'verified') await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
return {status:200, body:httpDeliveryReceipt(value, admitted.value)};
} catch (error) {
return {status:error instanceof HttpDeliveryError ? error.status : 503,
body:{error:error instanceof HttpDeliveryError ? error.code : 'delivery_unavailable'}};
} finally {
// Only authenticated deliveries are measured; a forged request logs nothing.
if (observed) emitGatewayDelivery({transport:'http', receivedAt, ...observed, outcome});
}
}
async admitGatewayDelivery(
delivery: Parameters<TagStateRpc['admitGatewayDelivery']>[0],
): ReturnType<TagStateRpc['admitGatewayDelivery']> {
const keyForm = await loadKeyFormFor(delivery);
const result = this.call((stores): GatewayAdmissionResult => {
if (parseHttpDeliveryState(stores.settings.getSetting(GATEWAY_HTTP_SETTING))?.mode === 'http') throw new Error('Socket delivery is disabled.');
const receipt = keyForm
? keyForm.keyFormReceiptNow(delivery, stores)
: gatewayUiReceiptNow(delivery, stores.uiSurfaces);
return receipt ?? stores.gatewayInbox.admit(delivery);
});
if (result.ok && typeof result.value === 'string') {
if (result.value === 'accepted') this.wakeRunnerAtReceipt(delivery);
await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
}
return result;
}
/**
* The direct path's receipt (CfSlackStateStore.noteDelivery): the Worker
* verified a direct installation's delivery and goes on to admit it over
* this store's RPCs. In one store turn, keep the receipt time for the turn
* admitted later (turnJobs.noteReceipt, so receiptToAdmissionMs means the
* same as on the gateway's path) and wake the runner the turn will be
* handed to (wakeRunnerAtReceipt); when one was predicted, the delivery
* becomes a turn whose status write needs the installation's identity, so
* start verifying it, held in this invocation (waitUntil) so the runtime
* carries the auth.test to its end, and the admission status write finds
* it fresh in this isolate (showStatusAtHandoff) instead of paying
* auth.test inside its cap. A delivery that wakes no runner (a join, an
* edit, a bot's post) verifies nothing. The Worker awaits none of it and
* nothing here changes what it admits: a failure fails nothing, and the
* identity is never logged.
*/
async noteSlackDelivery(note: SlackDeliveryNote): Promise<StateRpcResult<null>> {
const { workspaceId, eventId, eventTime, event, receivedAt } = note;
const noted = this.call((stores) => {
const installation = stores.config.getWorkspaceInstallation(workspaceId);
if (!installation || installation.transportMode !== 'direct' || installation.health === 'revoked') return undefined;
if (Number.isFinite(receivedAt)) stores.turnJobs.noteReceipt(eventId, receivedAt);
return this.createAlarmIdentityResolver(stores);
});
if (!noted.ok) return noted;
if (noted.value) {
const woken = this.wakeRunnerAtReceipt({ kind: 'event.deliver', envelope: { workspaceId, eventId, eventTime, event } });
if (woken) {
const verifying = noted.value(workspaceId).then(() => undefined, () => undefined);
// The runtime has waitUntil on DurableObjectState; the module declaration omits it.
const state = this.ctx as DurableObjectState & { waitUntil?(promise: Promise<unknown>): void };
if (typeof state.waitUntil === 'function') state.waitUntil(verifying);
}
}
return { ok: true, value: null };
}
/**
* CHICKPEA_SLACK_ADMISSION=store: admit a direct installation's verified
* message on this store's own stores. On arrival it checks the gate and the
* installation and takes its place in its thread's order; in that place it
* checks again and reads the credentials. Held by waitUntil; never throws.
*/
async admitSlackDelivery(delivery: SlackDirectDelivery): Promise<StateRpcResult<SlackDirectDeliveryResult>> {
const admitting = this.admitDirectDelivery(delivery);
const state = this.ctx as DurableObjectState & { waitUntil?(promise: Promise<unknown>): void };
if (typeof state.waitUntil === 'function') state.waitUntil(admitting);
return admitting;
}
private async admitDirectDelivery(delivery: SlackDirectDelivery): Promise<StateRpcResult<SlackDirectDeliveryResult>> {
const startedAt = Date.now();
const { workspaceId, eventId, eventTime, event, appId, receivedAt } = delivery;
const env = this.env as PlatformEnv;
const hosted = deploymentTenancy(env) === 'installation';
// Every store call admission makes is local; one over RPC would be this store calling itself.
const calls = new StateRpcSummary('store_admission');
const record = { turns: 0, orderWaitMs: 0, rateLimited: 0 };
const answer = (result: StateRpcResult<SlackDirectDeliveryResult>) => {
emitRuntimeLatency('slack_store_admission', {
outcome: result.ok ? result.value.outcome : 'failed',
reason: result.ok && result.value.outcome === 'refused' ? result.value.reason : undefined,
...record,
rpcCalls: calls.calls,
receiptToStartMs: Number.isFinite(receivedAt) ? startedAt - receivedAt : undefined,
durationMs: Date.now() - startedAt,
});
return result;
};
const refusal = (reason: 'recovery_only' | 'installation'): StateRpcResult<SlackDirectDeliveryResult> =>
({ ok: true, value: { outcome: 'refused', reason } });
const serving = (stores: TagStateStores) => {
if (stores.identity.getAuthControl()?.healthGate === 'recovery_only') return 'recovery_only' as const;
const installation = stores.config.getWorkspaceInstallation(workspaceId);
if (!installation || installation.transportMode !== 'direct' || installation.health === 'revoked' ||
(installation.appId && installation.appId !== appId) || (hosted && !installation.appId)) return 'installation' as const;
return installation;
};
const checked = this.call((stores) => {
const installation = serving(stores);
return typeof installation === 'string' ? installation : {
installation,
releaseReceipt: Number.isFinite(receivedAt) ? stores.turnJobs.noteReceipt(eventId, receivedAt) : undefined,
resolveIdentity: this.createAlarmIdentityResolver(stores),
};
});
if (!checked.ok) return answer(checked);
if (typeof checked.value === 'string') return answer(refusal(checked.value));
const { installation: arrived, releaseReceipt, resolveIdentity } = checked.value;
// In the check's synchronous turn: the thread's order is arrival order.
const place = this.directAdmissions.enter(slackEventConversationKey(event));
const enteredAt = Date.now();
const stores = this.stores!;
const handOffs: Array<Promise<void>> = [];
let result: StateRpcResult<SlackDirectDeliveryResult>;
try {
if (this.wakeRunnerAtReceipt({ kind: 'event.deliver', envelope: { workspaceId, eventId, eventTime, event } })) {
void resolveIdentity(workspaceId).catch(() => undefined);
}
result = await calls.run(async () => {
const local = localGatewayAppStores(stores);
if (hosted) await recordFirstHostedSlackDelivery(local.config, arrived);
await place;
record.orderWaitMs = Date.now() - enteredAt;
const current = this.call(serving);
if (!current.ok) return current;
if (typeof current.value === 'string') return refusal(current.value);
const installation = current.value;
let credentials: ResolvedSlackInstallationCredentials;
try {
credentials = await resolveSlackInstallationCredentials(
slackInstallationCredentialId(env), env, { state: stores.identity as unknown as IdentityStore, env },
);
} catch (error) {
if (error instanceof SlackCredentialRecoveryOnlyError) return refusal('recovery_only');
throw error;
}
const botUserId = credentials.botUserId ?? installation.botUserId ??
(await resolveIdentity(workspaceId).catch(() => undefined))?.botUserId;
const admitted = new Set<string>();
const slackState = this.admittingSlackState(local.slackState, (job) => {
admitted.add(job.id);
record.turns += 1;
// Armed at the commit: if this object stops before the hand-off, the alarm makes it.
handOffs.push(this.setAlarmNoLaterThan(Date.now() + KEEP_WARM_TICK_MS)
.then(() => this.afterSlackAdmission(job))
.catch(() => { console.warn('[chickpea] an admitted turn was not handed over; the alarm hands it over'); }));
});
await processDirectSlackEvent({
token: '', team_id: workspaceId, api_app_id: appId, event_id: eventId, event_time: eventTime,
type: 'event_callback', event,
} as SlackEventFixture, env, {
stores: { ...local, slackState },
credentials,
botUserId,
// As enqueueTurn: a turn admitted without a canonical admission (a legacy turn, an edit's).
enqueueTurn: async (job) => {
if (admitted.has(job.id)) return { ok: true, value: null };
stores.turnJobs.enqueue(job);
await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
return { ok: true, value: null };
},
onRateLimited: () => { record.rateLimited += 1; },
}, installation);
return { ok: true, value: { outcome: 'processed' } };
});
} catch (error) {
result = { ok: false, error: { code: 'internal', message: error instanceof Error ? error.message : String(error) } };
} finally {
// The thread's next delivery goes on beside this one's status and hand-off.
void place.then((release) => release());
// As managementExecute: Chickpea's introduction, queued by a person's first message, is sent now.
await this.armReceiptDrainAfter('claim_introduction').catch(() => undefined);
await Promise.all(handOffs);
releaseReceipt?.();
}
return answer(result);
}
/** This store's Slack state for admitSlackDelivery, doing what its RPCs do around the same writes. */
private admittingSlackState(local: SlackStateStore, admitted: (job: TurnJob) => void): SlackStateStore {
return {
...local,
admitCanonical: async (input) => {
const result = await local.admitCanonical(input);
if (result.claimed && 'steered' in result) await this.offerCreatedStop(result.steered);
if (input.turnJob && 'admission' in result) admitted(input.turnJob);
return result;
},
steerTurn: async (request, enqueue) => {
const decision = await local.steerTurn!(request, enqueue);
if (decision.outcome === 'enqueue' && decision.enqueued) await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
else await this.offerCreatedStop(decision);
return decision;
},
};
}
/**
* Wake the thread runner a just-received delivery's turn will be handed
* to, so that runner activates while the turn is admitted (see
* receiptRunnerWake for which deliveries). Never awaited: the gateway's
* acknowledgement waits for this receipt, and the direct path's note
* (noteSlackDelivery) is itself unawaited. Nothing here changes the receipt.
* Returns whether a runner was woken.
*/
private wakeRunnerAtReceipt(delivery: ReceiptWakeDelivery): boolean {
if (!this.runnerMode()) return false;
const predicted = this.call((stores) => receiptRunnerWake(delivery, stores.config));
return predicted.ok && predicted.value !== undefined && this.sendRunnerWake(predicted.value);
}
/**
* A thread runner's note that Chickpea's run decided to hand its root DM
* to an Agent (CfSlackStateStore.noteFrontDeskHandoff). That Agent's
* takeover turn is admitted about two seconds later, once the run has
* posted its line and settled. In one store turn, predict the runner the
* takeover will be handed to as admission will key it
* (frontDeskHandoffRunnerWake) and keep the decision time as that turn's
* receipt (turnJobs.noteReceipt, so its receiptToAdmissionMs and
* runnerWakeLeadMs measure from the decision); then wake the runner
* (sendRunnerWake). The runner awaits none of it, and nothing here changes
* what is admitted: a wrong prediction costs one activation.
*/
async noteFrontDeskHandoff(note: SlackFrontDeskHandoffNote): Promise<StateRpcResult<null>> {
const predicted = this.call((stores) => {
const wake = frontDeskHandoffRunnerWake(note, stores.config);
if (wake && Number.isFinite(note.decidedAt)) stores.turnJobs.noteReceipt(wake.eventId, note.decidedAt);
return wake;
});
if (!predicted.ok) return predicted;
if (predicted.value) this.sendRunnerWake(predicted.value);
return { ok: true, value: null };
}
/**
* Wake the runner at `threadKey`, naming the turn it will be handed, so it
* activates while that turn is admitted. Never awaited: the gateway's
* acknowledgement waits for its receipt, and the direct path's and the
* runner's notes are themselves unawaited. Returns whether a wake was sent.
*/
private sendRunnerWake({ threadKey, turnJobId }: ReceiptRunnerWake): boolean {
try {
if (!this.runnerMode()) return false;
const runner = threadRunnerStub(this.env as PlatformEnv, threadKey);
if (!runner) return false;
void Promise.resolve(runner.wake(Date.now(), turnJobId)).catch(() => undefined);
return true;
} catch {
// The hand-off addresses the runner either way.
return false;
}
}
async resumeTurnAfterOAuth(
originalTaskId: string,
continuationId: string,
): Promise<StateRpcResult<boolean>> {
const result = this.call((stores) => {
const resumed = stores.turnJobs.resumeAfterOAuth(originalTaskId, continuationId);
const id = oauthResumeTurnJobId(continuationId);
const view = resumed ? stores.turnJobs.runnerView(id) : undefined;
return {
resumed,
// A replayed callback for a continuation already handed to its
// thread runner re-admits it there (idempotent) rather than waiting
// for a sweep; a new continuation is dispatched like any turn.
runnerJob: view?.status === 'pending' && view.executor === 'runner' ? view.job : undefined,
};
});
if (!result.ok) return result;
const readmitted = result.value.runnerJob
? await this.admitToRunner(result.value.runnerJob)
: false;
if (result.value.resumed && !readmitted) {
await this.armAlarmNoLaterThan(this.admissionWakeAt(), true);
}
return { ok: true, value: result.value.resumed };
}
/**
* Cross-isolate activity narration (see src/slack/status-relay.ts): the agent
* DO observes safe lifecycle/tool summaries and relays them here, where the alarm
* registered the live turn's status presenter. A registry miss, closed sink,
* stale generation, or ambiguous duplicate match is intentionally a no-op —
* still a success by contract.
*/
async observedStatus(
instanceId: string,
submissionId: string,
status: TypedActivityStatus,
): Promise<StateRpcResult<null>> {
let runnerKey: string | undefined;
const result = this.call((stores) => {
if (!isSafeTypedActivityStatus(status)) return null;
const target = stores.turnJobs.matchFlueObservation(instanceId, submissionId);
if (target?.executor === 'runner') {
// A thread runner registered this turn's status; an agent without
// the dispatch's route in context still reaches it through here.
runnerKey = target.runnerKey;
} else if (target) {
defaultSlackStatusRegistry.setObservedStatus(
instanceId,
target.generation,
activityStatus(
status.kind,
status.action,
status.object,
status.family,
status.phase,
),
);
}
return null;
});
if (runnerKey) {
await threadRunnerStub(this.env as PlatformEnv, runnerKey)
?.observedStatus(instanceId, submissionId, status)
.catch(() => undefined);
}
return result;
}
/** A thread runner's turn reads, in one round trip (config/turn-read-batch.ts). */
async stateReadBatch(reads: readonly StateRead[]): Promise<Array<StateRpcResult<unknown> | null>> {
const { answerStateReads } = await import('./config/turn-read-batch.ts');
return answerStateReads(reads, this);
}
/** Per-turn writes from the SlackThreadRunner executing that turn. */
async threadRunnerTurn<K extends ThreadRunnerTurnKind>(
op: ThreadRunnerTurnOp<K>,
): Promise<StateRpcResult<ThreadRunnerTurnResult<K>>> {
// A runner learns from these that a code update replaced its version
// (see src/slack/thread-runner-loop.ts); reading it touches no storage.
const servingVersion = cloudflareWorkerVersionId(this.env);
if (op.kind === 'servingVersion') {
return { ok: true, value: (servingVersion ?? null) as ThreadRunnerTurnResult<K> };
}
const result = this.call((stores) =>
applyThreadRunnerTurnOp(stores, op as ThreadRunnerTurnOp, this.env as PlatformEnv) as ThreadRunnerTurnResult<K>);
// A turn's end renews the hold from it, as the runner's and the Agent
// object's count from it; for a person's turn only, as at admission.
if (result.ok && (op.kind === 'markDelivered' || op.kind === 'markError') && this.hold().active) {
const person = this.call((stores) => stores.turnJobs.isPersonTurn(op.id));
if (person.ok && person.value) this.hold().holdFrom(Date.now());
}
if (op.kind === 'begin' && result.ok && servingVersion !== undefined) {
return { ok: true, value: { ...(result.value as RunnerTurnBegin), servingVersion } as ThreadRunnerTurnResult<K> };
}
return result;
}
/**
* Drain queued turns past the events ack — the whole point of the relay. Each
* turn runs with this DO alarm's 15-minute wall-time budget instead of the
* events invocation's ~30s waitUntil cancellation, so a slow keyless model
* turn finishes and delivers.
*
* The handler NEVER throws for a per-job failure (it catches and either
* re-arms or gives up), so its attempt-count / delivered writes always commit
* on a normal return — no dependency on Durable Object throw-rollback
* semantics. It throws ONLY when the store itself is unavailable, so the
* platform's at-least-once alarm retry re-drives the queue after a transient
* storage error rather than dropping every job.
*
* Every invocation emits one content-free `relay_alarm` log (duration, jobs
* listed/run, groups, re-arm); emission cannot throw or change the result.
*/
async alarm(): Promise<void> {
this.boot.call('alarm');
const metrics = startRelayAlarmMetrics();
try {
await this.drainRelayAlarm(metrics);
} catch (error) {
metrics.outcome = 'threw';
throw error;
} finally {
emitRelayAlarm(metrics);
}
}
private async drainRelayAlarm(metrics: RelayAlarmMetrics): Promise<void> {
const alarmStartedAt = metrics.startedAt;
this.stores ??= this.tryInit();
if (!this.stores) {
throw new Error(`state store unavailable in alarm: ${this.initError ?? 'unknown'}`);
}
const stores = this.stores;
// A store woken cold learns from its newest turn row whether a
// conversation is live, and holds itself for the rest of it.
this.lastHumanTurnAt ??= stores.turnJobs.newestEnqueuedAt() ?? 0;
this.hold().holdFrom(this.lastHumanTurnAt);
const productTelemetry = createPlatformProductTelemetry({
env: this.env as PlatformEnv,
settings: localSettingsStore(stores),
config: localGatewayAppStores(stores).config,
});
// The retention sweep at most once per RETENTION_SWEEP_INTERVAL_MS; a
// skipped sweep reports 0 ms.
const retentionStartedAt = Date.now();
let retentionMs = 0;
if (this.lastRetentionSweepAt === undefined ||
retentionStartedAt - this.lastRetentionSweepAt >= RETENTION_SWEEP_INTERVAL_MS) {
stores.management.cleanupRetention(retentionStartedAt, 250);
this.lastRetentionSweepAt = retentionStartedAt;
retentionMs = Date.now() - retentionStartedAt;
}
// A practice run goes an hour after it settles, on the wake nextDueAt arms for it.
stores.testRuns.eraseExpired(retentionStartedAt);
// New turns go to their thread's SlackThreadRunner (the default executor)
// and this alarm only finishes turns it already dispatched to Flue: rows
// admitted before the runner became the default. The emergency gate
// SLACK_TAG_TURN_EXECUTOR=alarm (or a Worker without the runner binding)
// keeps every new turn on this alarm instead. Rows handed to runners stay
// theirs either way. The alarm execution path below is kept for those
// legacy and fallback rows; delete it once a release has drained them.
const runnerBinding = Boolean((this.env as PlatformEnv).SLACK_THREAD_RUNNER);
const runnerMode = runnerBinding &&
slackTurnExecutor(this.env as PlatformEnv) === 'runner';
const handingOff = new Set<Promise<void>>();
/**
* Hand the free turns listed now to their runners. Each admission starts
* its own hand-off at once, side by side with any in flight: a row is
* assigned before its admission RPC is awaited, so two hand-offs never
* take the same row. A failed hand-off stays a hand-off; the alarm
* re-arms for it.
*/
const handOff = (readmitHandoffs: boolean): Promise<void> => {
// Unconfirmed hand-offs are admitted again even after the switch is
// turned off: those rows already belong to their runners. Only when no
// hand-off is in flight, so a row mid-admission is not admitted twice.
if (!runnerMode && !(runnerBinding && stores.turnJobs.hasHandoffs())) return Promise.resolve();
const readmit = readmitHandoffs && handingOff.size === 0;
const run: Promise<void> = this.dispatchToRunners(stores, runnerMode, readmit)
.then((dispatched) => { metrics.jobsDispatched += dispatched; }, () => undefined)
.finally(() => { handingOff.delete(run); });
handingOff.add(run);
return run;
};
/** Hand over every free turn and wait for all hand-offs in flight. */
const dispatchToRunners = async (): Promise<void> => {
await handOff(true);
while (handingOff.size > 0) await Promise.all([...handingOff]);
};
const onAdmitted = runnerMode ? () => void handOff(false) : undefined;
/** Admit newly delivered events and hand their turns over at once. */
const admitAndDispatch = async () => {
let retry = false;
// Runner mode: an admission that lands while this pass runs (before any
// wake is listening) is picked up by another pass, not the next alarm.
for (let pass = 0; pass < 8; pass += 1) {
const admissions = this.admissionsSeen;
retry = (await drainGatewayInbox(stores, this.env as PlatformEnv, onAdmitted)) || retry;
await dispatchToRunners();
if (!runnerMode || this.admissionsSeen === admissions) break;
}
return retry;
};
/**
* Runner mode: the alarm's other due work (ledger runs, cleanups,
* repairs, schedule actions, receipts) can take seconds. Keep admitting
* and handing over turns that arrive meanwhile, on every admission wake
* and at the admission re-check interval, instead of after it.
*/
const whileDispatching = <T>(work: () => Promise<T>): Promise<T> =>
runnerMode ? this.dispatchingWhile(work, admitAndDispatch) : work();
// Test runs settle and erase beside everything else; a step waits on its instance, never on a turn.
const testRunPass = stores.testRuns.hasUnerased()
? this.testRuns().then((lifecycle) => lifecycle.pass()).catch((error: unknown) => {
console.warn('[chickpea] test run pass failed', error instanceof Error ? error.name : typeof error);
})
: undefined;
const gatewayNeedsRetry = await admitAndDispatch();
// Stops owed to runners, after the hand-offs above gave new rows theirs.
await this.deliverStopNotices();
const threadKeyOf = (job: {
turn: Parameters<typeof slackAgentThreadKey>[0];
assignment: Parameters<typeof slackAgentThreadKey>[1];
}) => slackAgentThreadKey(job.turn, job.assignment);
const listPendingTurns = () => stores.turnJobs.listPendingByThread({
maxThreads: MAX_TURN_DRAIN_BATCH,
perThread: ALARM_PENDING_PER_THREAD,
threadKey: threadKeyOf,
executor: 'alarm',
dispatchedOnly: runnerMode,
});
const carriedJobIds = () => new Set(this.carriedAlarmTurns.keys());
const pending = listPendingTurns();
metrics.jobsListed = pending.length;
if (pending.length === 0) {
const resolveInstallation = this.createAlarmIdentityResolver(stores);
/** Runs one part of the idle drain and records how long it took. */
const timed = async <T>(
part: 'ledgerMs' | 'cleanupsMs' | 'repairsMs' | 'scheduleMs' | 'receiptsMs',
work: () => Promise<T>,
): Promise<T> => {
const startedAt = Date.now();
try {
return await work();
} finally {
metrics[part] = Date.now() - startedAt;
}
};
metrics.retentionMs = retentionMs;
const { ledgerDrain, presentationRepairs, scheduleActions, cleanupPending } = await whileDispatching(async () => {
const ledgerDrain = await timed('ledgerMs', () => drainLedgerRuns(
stores,
this.env as PlatformEnv,
resolveInstallation,
productTelemetry,
));
// One scan per firing: after the ledger drain (a ledger run ending
// here can leave a receipt to clean), reused by the retry below.
const cleanupPending = stores.turnJobs.hasPendingSlackInteractionCleanup();
if (cleanupPending) {
await timed('cleanupsMs', () => drainSlackInteractionCleanups(stores, resolveInstallation, carriedJobIds()));
} else {
metrics.cleanupsMs = 0;
}
const presentationRepairs = await timed('repairsMs', () => drainTerminalPresentationRepairs(
stores,
this.env as PlatformEnv,
resolveInstallation,
carriedJobIds(),
));
const scheduleActions = await timed('scheduleMs', () =>
drainCloudflareScheduleActions(stores, this.env as PlatformEnv));
await timed('receiptsMs', () =>
drainCloudflareManagementReceipts(stores, this.env as PlatformEnv, resolveInstallation, this.presentationRunnerOf));
return { ledgerDrain, presentationRepairs, scheduleActions, cleanupPending };
});
const turnRetry = gatewayNeedsRetry || stores.gatewayInbox.hasPending() ||
stores.turnJobs.hasPending('ledger') || cleanupPending ||
stores.turnJobs.hasHandoffs() ||
(runnerMode && stores.turnJobs.hasPending('legacy'))
? Date.now() + runDriverRetryDelayMs(ledgerDrain, RELAY_RETRY_BACKOFF_MS)
: undefined;
const outboxRetry = stores.management.nextOutboxDueAt();
const keepWarmAt = this.keepWarmTickAt();
// An instance that does not answer costs a step its bounds; turns keep being handed over meanwhile.
if (testRunPass) await whileDispatching(() => testRunPass);
const nextWake = earliestDefined(
turnRetry,
presentationRepairs.nextRetryAt,
scheduleActions.nextDueAt,
outboxRetry,
stores.gatewayInbox.nextPendingDueAt(),
keepWarmAt,
stores.testRuns.nextDueAt(Date.now()),
);
metrics.needsRetry = turnRetry !== undefined;
metrics.rearmed = nextWake !== undefined;
metrics.keepWarm = keepWarmAt !== undefined;
if (nextWake !== undefined) await this.armAlarmNoLaterThan(nextWake);
return;
}
metrics.outcome = 'drained';
// Resolve current credentials once per identity referenced by this bounded
// batch. The map is discarded after the alarm, so the next retry observes
// credential rotation without ever falling back to another identity.
const resolveInstallation = this.createAlarmIdentityResolver(stores);
const usageStore = localUsageStore(stores);
const appStores = localGatewayAppStores(stores);
const resolveManagementApproval = () => {
const managementRuntime = localManagementRuntime(
stores,
this.env as PlatformEnv,
appStores,
);
return {
identity: appStores.identity,
config: appStores.config,
management: appStores.management,
service: managementRuntime.service,
};
};
let needsRetry = gatewayNeedsRetry;
let identityRetryDelayMs = RELAY_RETRY_BACKOFF_MS;
// The alarm's own stores are this turn's ports; a thread runner passes
// its own presentation state (see src/slack/turn-executor.ts).
const turnPorts: TurnExecutionPorts = {
env: this.env as PlatformEnv,
turnJobs: alarmTurnJobsPort(stores),
slack: stores.slack,
config: stores.config,
presentationState: localSlackPresentationState(stores, this.env as PlatformEnv),
settingsStore: localSettingsStore(stores),
usageStore,
workStore: stores.work as unknown as WorkStore,
appStores,
managementApproval: resolveManagementApproval,
telemetry: productTelemetry,
resolveInstallation,
sandboxes: sandboxTurnReaders(this.env as PlatformEnv),
dispatchAgentAsks: localAgentAskDispatcher(stores, this.env as PlatformEnv, productTelemetry),
dispatchFrontDeskHandoff: localFrontDeskDispatcher(stores, this.env as PlatformEnv, productTelemetry),
runTurn,
};
const runJob = (
job: (typeof pending)[number],
control?: AlarmTurnJobControl,
): Promise<boolean> => executeTurnJob(job, turnPorts, {
latency: { lane: 'cloudflare', executor: 'alarm' },
...(control ? { control } : {}),
// The alarm executor has no coding cascade; its stop note reads only
// whether a stop was recorded (R22 versus the stopped ending).
stopRecorded: () => stores.turnJobs.runnerView(job.id).job?.stop?.role === 'stopped',
onRetry: (afterMs) => {
needsRetry = true;
if (afterMs !== undefined) {
identityRetryDelayMs = Math.max(identityRetryDelayMs, afterMs);
}
},
});
// Ordering inside a thread is preserved (a thread's second turn never
// overtakes its first) while unrelated conversations run side by side.
// The drain keeps admitting while turns observe, so one long turn never
// holds new messages until the platform ends this alarm, and it yields
// every observation before the alarm's wall-time limit (see
// src/slack/alarm-turn-drain.ts). Only the thread cap bounds it.
const drainedThreads = new Set<string>();
const turnsStartedAt = Date.now();
const turnDrain = await drainAlarmTurnJobs<(typeof pending)[number]>({
initial: pending,
jobId: (job) => job.id,
threadKey: threadKeyOf,
runJob: async (job, control) => {
// Never while a stop's abort for an earlier turn of the thread is out.
await this.stopAborts.clear(threadKeyOf(job));
const jobStartedAt = Date.now();
metrics.jobsRun += 1;
drainedThreads.add(threadKeyOf(job));
const settled = await runJob(job, control);
// One attempt: a yielded turn's next attempt is a later alarm's job.
metrics.longestJobMs = Math.max(metrics.longestJobMs, Date.now() - jobStartedAt);
if (settled) metrics.jobsSettled += 1;
else metrics.jobsRetained += 1;
return settled;
},
refresh: async () => {
needsRetry = (await admitAndDispatch()) || needsRetry;
return listPendingTurns();
},
carried: {
threadKeys: () => new Set(this.carriedAlarmTurns.values()),
jobIds: carriedJobIds,
},
// Due receipts, schedule actions, repairs, and cleanups keep moving
// while a long turn observes. Records owned by a running job are its
// own to finish; the tail picks them up once it returns.
tick: async (runningJobIds) => {
await drainSlackInteractionCleanups(stores, resolveInstallation, runningJobIds);
await drainTerminalPresentationRepairs(stores, this.env as PlatformEnv, resolveInstallation, runningJobIds);
await drainCloudflareScheduleActions(stores, this.env as PlatformEnv);
await drainCloudflareManagementReceipts(stores, this.env as PlatformEnv, resolveInstallation, this.presentationRunnerOf);
},
startConcurrency: MAX_TURN_DRAIN_BATCH,
maxActiveThreads: MAX_TURN_DRAIN_BATCH,
startedAt: alarmStartedAt,
budgetMs: ALARM_TURN_BUDGET_MS,
hardCapMs: ALARM_TURN_HARD_CAP_MS,
recheckMs: ALARM_ADMISSION_RECHECK_MS,
onWake: (wake) => {
this.alarmAdmissionWake = wake;
return () => {
if (this.alarmAdmissionWake === wake) this.alarmAdmissionWake = undefined;
};
},
});
metrics.groups = drainedThreads.size;
metrics.turnsMs = Date.now() - turnsStartedAt;
metrics.yielded = turnDrain.budgetExhausted;
metrics.jobsCarried = turnDrain.carried.length;
for (const carried of turnDrain.carried) {
console.warn('[chickpea] Turn job still running at the alarm cap; its thread waits for it');
this.carriedAlarmTurns.set(carried.id, carried.threadKey);
void carried.settled.finally(() => {
this.carriedAlarmTurns.delete(carried.id);
// Its thread may continue now; a running drain picks that up at once.
this.alarmAdmissionWake?.();
});
}
const { presentationRepairs, scheduleActions } = await whileDispatching(async () => {
const ledgerDrain = await drainLedgerRuns(
stores,
this.env as PlatformEnv,
resolveInstallation,
productTelemetry,
);
identityRetryDelayMs = runDriverRetryDelayMs(ledgerDrain, identityRetryDelayMs);
await drainSlackInteractionCleanups(stores, resolveInstallation, carriedJobIds());
const presentationRepairs = await drainTerminalPresentationRepairs(
stores,
this.env as PlatformEnv,
resolveInstallation,
carriedJobIds(),
);
const scheduleActions = await drainCloudflareScheduleActions(stores, this.env as PlatformEnv);
await drainCloudflareManagementReceipts(stores, this.env as PlatformEnv, resolveInstallation, this.presentationRunnerOf);
return { presentationRepairs, scheduleActions };
});
needsRetry ||= stores.turnJobs.hasPending('legacy') ||
stores.turnJobs.hasPending('ledger') ||
stores.turnJobs.hasPendingSlackInteractionCleanup() ||
stores.gatewayInbox.hasPending() ||
stores.turnJobs.hasHandoffs();
// Yielded turns are still running elsewhere; reattach to them promptly.
const turnRetry = turnDrain.budgetExhausted
? Date.now() + ALARM_YIELD_REARM_MS
: needsRetry ? Date.now() + identityRetryDelayMs : undefined;
const outboxRetry = stores.management.nextOutboxDueAt();
if (testRunPass) await whileDispatching(() => testRunPass);
const nextWake = earliestDefined(
turnRetry,
presentationRepairs.nextRetryAt,
scheduleActions.nextDueAt,
outboxRetry,
stores.gatewayInbox.nextPendingDueAt(),
this.keepWarmTickAt(),
stores.testRuns.nextDueAt(Date.now()),
);
metrics.needsRetry = needsRetry;
metrics.rearmed = nextWake !== undefined;
if (nextWake !== undefined) {
// Re-arm (do NOT throw) so this invocation returns normally and its
// attempt-count writes commit; the next firing re-drives the leftover
// pending jobs. Preserve an earlier wake armed by an RPC while this
// drain was awaiting external I/O.
await this.armAlarmNoLaterThan(nextWake);
}
}
/**
* Admit one runner-owned turn to its thread's SlackThreadRunner, with the
* presentation as this store holds it, and confirm the hand-off. Idempotent:
* a runner keeps a job it already holds. False leaves the row a hand-off for
* the next alarm to admit again. A turn's first hand-off may show its status
* first (showStatusAtHandoff); a hand-off made again never does.
*/
private async admitToRunner(job: PendingTurnJob, firstHandoff = false): Promise<boolean> {
const stores = this.stores;
const threadKey = slackAgentThreadKey(job.turn, job.assignment);
const runner = threadRunnerStub(this.env as PlatformEnv, threadKey);
if (!stores || !runner) return false;
const slackInstallation = stores.config.getWorkspaceInstallation(effectiveTurnSlackInstallationId(job.turn));
// The bot identity this store verified within the window, so the runner
// does not ask Slack for it again before the turn's first status.
const gatewayIdentity = carriedGatewayIdentity(slackInstallation);
const installation = installationOwnershipOf(this.env as PlatformEnv);
const agentWake = this.wakeAgentAtHandoff(job, {
installation, botUserId: slackInstallation?.botUserId, inventory: stores.objectInventory,
});
let admissionStatus: AdmissionStatusMark | undefined;
if (firstHandoff) {
this.firstHandoffs.add(job.id);
try {
admissionStatus = await this.showStatusAtHandoff(job, slackInstallation?.botUserId);
} finally {
this.firstHandoffs.delete(job.id);
}
}
// Read after the status: the runner's copy is the only one from here on.
const presentation = job.runId ? stores.presentations.get(job.runId) : undefined;
const payload: ThreadRunnerJobPayload = {
...(presentation ? { presentation } : {}),
...(gatewayIdentity ? { gatewayIdentity } : {}),
...(agentWake ? { agentWake } : {}),
...(admissionStatus ? { admissionStatus } : {}),
};
try {
const result = await runner.admit({
id: job.id,
threadKey,
payload,
...(installation ? { installation } : {}),
sentAt: Date.now(),
});
if (result.refused) throw new Error(result.refused);
} catch {
console.warn('[chickpea] Thread runner admission failed; the hand-off is retried');
return false;
}
stores.turnJobs.confirmRunner(job.id);
return true;
}
/**
* Wake the Flue agent object a handed-over turn's plan will name, so its
* activation overlaps the hand-off, the runner's start and the turn's
* preparation instead of starting once the plan froze. The name is
* recorded in the inventory before it is addressed (admissionAgentInstance).
* Sent before the hand-off and never awaited; nothing here changes the
* hand-off. Returns what the hand-off carries, so the turn's freeze sends
* no second wake to the same object and measures its lead from this one.
*/
private wakeAgentAtHandoff(job: PendingTurnJob, facts: AdmissionAgentFacts): AdmissionAgentWake | undefined {
try {
const instanceId = admissionAgentInstance(job, facts);
if (!instanceId) return undefined;
const sentAt = Date.now();
void wakeSlackAgentObject(this.env as PlatformEnv, CHICKPEA_SLACK_AGENT_BINDING, instanceId, job.id, sentAt)
.catch(() => undefined);
return { instanceId, sentAt };
} catch {
// The freeze wakes the object instead.
return undefined;
}
}
/**
* Show a turn's opening status as its Agent before its first hand-off,
* instead of at the runner's start: only on a turn that will dispatch, has
* nothing ahead of it in its conversation (a turn waiting behind another
* must not replace that turn's status), and is not an Agent with its own
* Slack app (admissionStatusPresentation). The hand-off waits at most
* ADMISSION_STATUS_CAP_MS. The row records the status only if Slack
* acknowledged it in time and nothing else moved the row meanwhile; the
* runner otherwise shows it as before. Never fails the hand-off.
*/
private async showStatusAtHandoff(
job: PendingTurnJob,
botUserId: string | undefined,
): Promise<AdmissionStatusMark | undefined> {
try {
const stores = this.stores;
if (!stores || !job.runId) return undefined;
const presentation = admissionStatusPresentation(job, stores.presentations.get(job.runId), botUserId);
if (!presentation || stores.turnJobs.queuedBehind(job.id)) return undefined;
const resolveInstallation = this.createAlarmIdentityResolver(stores);
return await showAdmissionStatus({
presentation,
agentId: job.assignment.agent.id,
state: localSlackPresentationState(stores, this.env as PlatformEnv),
client: async () => (await resolveInstallation(effectiveTurnSlackInstallationId(job.turn))).client,
commit: (shown) =>
stores.presentations.get(shown.runId)?.projectionVersion === presentation.projectionVersion &&
stores.presentations.putSnapshot(shown),
});
} catch {
return undefined;
}
}
/**
* The default executor: hand every pending turn whose thread is
* free to its runner, a page at a time, and return without waiting for any
* turn. Unconfirmed hand-offs are admitted again first, so a hand-off lost
* to a restart of this object is never overtaken in its thread. A failed
* admission leaves its row a hand-off, which holds only its own thread
* (listDispatchable) and re-arms this alarm (hasHandoffs). Returns the
* turns admitted.
*/
private async dispatchToRunners(
stores: TagStateStores,
newTurns: boolean,
readmitHandoffs = true,
): Promise<number> {
let dispatched = 0;
const admitAll = async (jobs: PendingTurnJob[], firstHandoff: boolean) => {
const admitted = await Promise.all(jobs.map((job) => this.admitToRunner(job, firstHandoff)));
dispatched += admitted.filter(Boolean).length;
};
if (readmitHandoffs) {
await admitAll(stores.turnJobs.listHandoffs(MAX_TURN_DRAIN_BATCH)
.filter((job) => !this.firstHandoffs.has(job.id)), false);
}
for (let page = 0; newTurns && page < RUNNER_DISPATCH_MAX_PAGES; page += 1) {
const jobs = stores.turnJobs.listDispatchable({
limit: MAX_TURN_DRAIN_BATCH,
threadKey: (job) => slackAgentThreadKey(job.turn, job.assignment),
}).filter((job) => stores.turnJobs.assignRunner(job.id));
if (jobs.length === 0) break;
await admitAll(jobs, true);
}
return dispatched;
}
/**
* Run the alarm's other work while admitting and handing over turns that
* arrive meanwhile: on every admission wake, at the admission re-check
* interval, and once more after the work if anything was admitted during
* the last pass.
*/
private async dispatchingWhile<T>(
work: () => Promise<T>,
admitAndDispatch: () => Promise<unknown>,
): Promise<T> {
let done = false;
let wake: (() => void) | undefined;
let admittedMeanwhile = false;
this.dispatchWake = () => {
admittedMeanwhile = true;
wake?.();
};
const result = work().finally(() => {
done = true;
wake?.();
});
const loop = (async () => {
while (!done) {
await new Promise<void>((resolve) => {
const timer = setTimeout(resolve, ALARM_ADMISSION_RECHECK_MS);
wake = () => {
clearTimeout(timer);
resolve();
};
});
wake = undefined;
if (done) break;
admittedMeanwhile = false;
await admitAndDispatch();
}
if (admittedMeanwhile) await admitAndDispatch();
})();
try {
return await result;
} finally {
await loop.catch(() => undefined);
this.dispatchWake = undefined;
}
}
private admissionWakeAt(): number {
return this.runnerMode() ? Date.now() : Date.now() + RELAY_BATCH_WINDOW_MS;
}
private runnerMode(): boolean {
return Boolean((this.env as PlatformEnv).SLACK_THREAD_RUNNER) &&
slackTurnExecutor(this.env as PlatformEnv) === 'runner';
}
/**
* This store's residency hold, for the window the deployment sets
* (RESIDENCY_HOLD_VARIABLE): from a person's admitted turn, and again from
* its end, which the runner reports (threadRunnerTurn).
*/
private hold(): ResidencyHold {
this.conversationHold ??= new ResidencyHold(residencyHoldMs(this.env as PlatformEnv));
return this.conversationHold;
}
/**
* The next keep-warm tick while a person's turn was admitted here within
* KEEP_WARM_WINDOW_MS; undefined outside the window, and before any turn is
* known. Armed as the latest the alarm may wait (armAlarmNoLaterThan keeps
* an earlier due time), so a firing inside the window re-arms it. A
* residency hold stops the store hibernating; the tick alone does not.
* Where a pending timer does not block eviction (see ResidencyHold), each
* firing restarts the 70 to 140 seconds an idle object has before it.
*/
private keepWarmTickAt(): number | undefined {
const now = Date.now();
return this.lastHumanTurnAt !== undefined && now < this.lastHumanTurnAt + KEEP_WARM_WINDOW_MS
? now + KEEP_WARM_TICK_MS
: undefined;
}
private async armAlarmNoLaterThan(at: number, admission = false): Promise<void> {
// Every admission arms the alarm after its durable write. A running alarm
// cannot be re-entered, so let its drain pick the new work up directly.
this.alarmAdmissionWake?.();
if (admission) this.admissionsSeen += 1;
this.dispatchWake?.();
await this.setAlarmNoLaterThan(at);
}
/** Arms the alarm at `at` unless an earlier one is set; wakes nothing. */
private async setAlarmNoLaterThan(at: number): Promise<void> {
const existing = await this.ctx.storage.getAlarm();
if (existing === null || at < existing) await this.ctx.storage.setAlarm(at);
}
private createAlarmIdentityResolver(stores: TagStateStores): SlackInstallationExecutionResolver {
return withAgentAppExecution(cacheSlackInstallationExecutionContexts(
(workspaceId) => resolveSlackInstallationExecutionContext(
workspaceId,
this.env as PlatformEnv,
{
config: {
getWorkspaceInstallation: async (workspaceId) =>
stores.config.getWorkspaceInstallation(workspaceId),
},
settings: {
getSetting: async (key) => stores.settings.getSetting(key),
getSettings: async (keys) => stores.settings.getSettings(keys),
setSetting: async (key, value) => stores.settings.setSetting(key, value),
deleteSetting: async (key) => stores.settings.deleteSetting(key),
applySettingsPatch: async (patch) => stores.settings.applySettingsPatch(patch),
mergeSettingStringSet: async (key, values) =>
stores.settings.mergeSettingStringSet(key, values),
},
credentialDependencies: {
// The alarm is already executing inside TAG_STATE; using the
// local logic avoids a self-RPC while retaining durable state.
state: stores.identity as unknown as IdentityStore,
env: this.env as PlatformEnv,
},
},
),
), this.env as PlatformEnv);
}
/**
* Run one store operation and map the outcome onto the RPC envelope. Typed
* domain errors become stable codes with their constructor args so the
* proxies (cf-state-proxies.ts) re-throw the SAME instanceof-able errors the
* node backend throws; anything else is an internal failure with the message
* preserved for server-side logs.
*/
private call<T>(fn: (stores: TagStateStores) => T): StateRpcResult<T> {
this.boot.call('rpc');
// Self-heal: re-attempt a construction that failed on first boot rather
// than latching the isolate into permanent failure. A still-broken store
// returns the {ok:false} envelope only for THIS call.
this.stores ??= this.tryInit();
if (!this.stores) {
return {
ok: false,
error: {
code: 'internal',
message: `state store unavailable: init failed (${this.initError ?? 'unknown'})`,
},
};
}
try {
return { ok: true, value: fn(this.stores) };
} catch (err) {
if (err instanceof UnknownAgentError) {
return rpcError('unknown_agent', err.message, { agentId: err.agentId });
}
if (err instanceof AgentExistsError) {
return rpcError('agent_exists', err.message, { agentId: err.agentId });
}
if (err instanceof AgentRevisionConflictError) {
return rpcError('agent_revision_conflict', err.message, {
agentId: err.agentId,
expectedRevision: String(err.expectedRevision),
actualRevision: String(err.actualRevision),
});
}
if (err instanceof AgentConnectionCredentialConflictError) {
return rpcError('agent_connection_credential_conflict', err.message, {
agentId: err.agentId,
lane: err.lane,
connectionId: err.connectionId,
});
}
if (err instanceof ReservedAgentIdentityError) {
return rpcError('reserved_agent_identity', err.message, { field: err.field });
}
if (err instanceof WorkspaceModelDefaultRevisionConflictError) {
return rpcError('workspace_model_default_revision_conflict', err.message, {
workspaceId: err.workspaceId,
expectedRevision: String(err.expectedRevision),
actualRevision: String(err.actualRevision),
});
}
if (err instanceof ModelRoleRevisionConflictError) {
return rpcError('model_role_revision_conflict', err.message, {
scope: err.scope,
targetId: err.targetId,
role: err.role,
expectedRevision: String(err.expectedRevision),
actualRevision: String(err.actualRevision),
});
}
if (err instanceof AgentStillAssignedError) {
return rpcError('agent_still_assigned', err.message, {
agentId: err.agentId,
keys: err.keys,
});
}
if (err instanceof AgentStillReferencedError) {
return rpcError('agent_still_referenced', err.message, {
agentId: err.agentId,
references: err.references,
});
}
if (err instanceof ChannelRevisionConflictError) {
return rpcError('channel_revision_conflict', err.message, {
workspaceId: err.workspaceId,
channelId: err.channelId,
expectedRevision: String(err.expectedRevision),
actualRevision: String(err.actualRevision),
});
}
if (err instanceof ConnectionAccountRevisionConflictError) {
return rpcError('connection_account_revision_conflict', err.message, {
accountId: err.accountId,
expectedRevision: String(err.expectedRevision),
actualRevision: String(err.actualRevision),
});
}
if (err instanceof ConnectionAccountAlreadyBoundError) {
return rpcError('connection_account_already_bound', err.message, {
accountId: err.accountId,
agentId: err.agentId,
});
}
if (err instanceof ManagedRemoteAccountAlreadyUsedError) {
return rpcError('managed_remote_account_already_used', err.message, {
adapterId: err.adapterId,
accountRef: err.accountRef,
});
}
if (err instanceof ManagedSignInGroupConflictError) {
return rpcError('managed_sign_in_group_conflict', err.message, {
adapterId: err.adapterId,
accountRef: err.accountRef,
});
}
if (err instanceof IdentityStateError) {
return rpcError('identity', err.message, {
identityCode: err.code,
...err.details,
});
}
if (err instanceof ManagementError) {
return rpcError('management', err.message, {
managementCode: err.code,
});
}
if (err instanceof MemoryStateError) {
return rpcError('memory', err.message, {
memoryCode: err.code,
...err.details,
});
}
if (err instanceof RoutineStateError) {
return rpcError('routine', err.message, {
routineCode: err.code,
...err.details,
});
}
if (err instanceof UsageStateError) {
return rpcError('usage', err.message, {
usageCode: err.code,
...err.details,
});
}
if (err instanceof WorkStateError) {
return rpcError('work', err.message, {
workCode: err.code,
...err.details,
});
}
if (err instanceof SlackPresentationStateError) {
return rpcError('slack_presentation', err.message, {
presentationCode: err.code,
});
}
const message = err instanceof Error ? err.message : String(err);
console.error('[chickpea] TagStateStore RPC failure:', message);
return rpcError('internal', message);
}
}
}
function runnerSlackCredentials(
stores: TagStateStores,
env: PlatformEnv,
): Pick<RunnerTurnBegin, 'slackCredentials'> {
const identityId = slackInstallationCredentialId(env);
try {
return {
slackCredentials: {
identityId,
control: stores.identity.getSlackCredentialControl() ?? null,
active: stores.identity.getActiveSlackCredentialRevision(identityId) ?? null,
},
};
} catch {
return {};
}
}
/** One per-turn operation of a SlackThreadRunner (see thread-runner-rpc.ts). */
function applyThreadRunnerTurnOp(
stores: TagStateStores,
op: ThreadRunnerTurnOp,
env: PlatformEnv,
): ThreadRunnerTurnResult<ThreadRunnerTurnKind> {
const turnJobs = stores.turnJobs;
switch (op.kind) {
case 'view':
return turnJobs.runnerView(op.id);
case 'begin': {
const view = turnJobs.runnerView(op.id);
const installation = view.job
? stores.config.getWorkspaceInstallation(view.job.turn.workspaceId)
: undefined;
const keys = runnerPrefetchedSettings();
const values = stores.settings.getSettings(keys);
const presentation = view.job?.runId ? stores.presentations.get(view.job.runId) : undefined;
return {
view,
...(installation ? { installation } : {}),
...(installation?.transportMode === 'direct' ? runnerSlackCredentials(stores, env) : {}),
...(presentation?.schemaVersion === 3
? {
latestThreadSessionGeneration:
stores.presentations.getLatestThreadSessionGeneration(presentation.root) ?? null,
}
: {}),
settings: Object.fromEntries(
keys.map((key, index) => [key, values[index] ?? null]),
),
};
}
case 'recordAttempt':
turnJobs.recordAttempt(op.id, op.attempts);
return null;
case 'recordPullRequest':
return turnJobs.recordPullRequest(op.id, op.pullRequest) ?? null;
case 'freezeRuntimePlan':
return turnJobs.freezeRuntimePlan(op.id, op.candidate);
case 'getBoundRuntimePlan':
return turnJobs.getBoundRuntimePlan(
op.continuityKey,
op.beforeMessageTs,
op.actorMembershipId,
op.agentId,
) ?? null;
case 'getThreadContinuation':
return turnJobs.getThreadContinuation(op.continuityKey, op.instanceId, op.beforeMessageTs) ?? null;
case 'fold':
return turnJobs.fold(op.request);
case 'recordUsagePersistence':
return turnJobs.recordUsagePersistence(op.id, op.event) ?? null;
case 'recordInteractionIntent':
return turnJobs.recordInteractionIntent(op.id, op.intent) ?? null;
case 'markDelivered':
return stores.slack.endTurn(op.id, turnJobs, stores.work, stores.presentations);
case 'markError':
turnJobs.markError(op.id);
return null;
case 'markCodingActiveWork':
stores.slack.markCodingActiveWork(op.key, op.generation);
return null;
case 'isCodingActiveWork':
return stores.slack.isCodingActiveWork(op.key, op.generation);
case 'putPresentation':
return stores.presentations.putSnapshot(op.presentation);
case 'servingVersion':
// Answered by TagStateStore.threadRunnerTurn without storage.
return null;
default:
throw new Error('Unknown thread runner operation.');
}
}
async function drainSlackInteractionCleanups(
stores: TagStateStores,
resolveInstallation: SlackInstallationExecutionResolver,
excludeTurnJobIds: ReadonlySet<string> = new Set(),
): Promise<void> {
for (const job of stores.turnJobs.listPendingSlackInteractionCleanups(MAX_TURN_DRAIN_BATCH)) {
if (excludeTurnJobIds.has(job.id)) continue;
const progress = job.progress.slackInteraction;
if (!progress) continue;
try {
const installationContext = await resolveInstallation(effectiveTurnSlackInstallationId(job.turn));
await verifySlackInstallationTurnAccess(installationContext, job.turn);
await repairSlackInteractionProgress(
job.turn,
job.assignment,
progress,
installationContext.client,
(patch) => {
stores.turnJobs.recordSlackInteractionProgress(job.id, patch);
},
);
} catch (error) {
console.warn('[chickpea] Slack interaction cleanup retry failed:', sanitizeError(error));
}
}
}
async function drainTerminalPresentationRepairs(
stores: TagStateStores,
env: PlatformEnv,
resolveInstallation: SlackInstallationExecutionResolver,
excludeTurnJobIds: ReadonlySet<string> = new Set(),
): Promise<SlackPresentationRepairDrainResult> {
// A thread runner repairs the presentations of turns it executes.
const presentations = stores.presentations
.listAutoRepairableV3(MAX_TURN_DRAIN_BATCH, { skipRunnerOwned: true })
.filter((presentation) => !excludeTurnJobIds.has(presentation.turnJobId));
return drainSlackPresentationRepairs({
presentations,
state: localSlackPresentationState(stores, env),
resolveClient: async (workspaceId) => (await resolveInstallation(workspaceId)).client,
onFailure: (_presentation, error) => {
console.warn('[chickpea] Slack presentation repair failed:', sanitizeError(error));
},
});
}
async function drainCloudflareManagementReceipts(
stores: TagStateStores,
env: PlatformEnv,
resolveInstallation: SlackInstallationExecutionResolver,
presentationRunner: (runId: string) => SlackThreadRunnerRpc | undefined = () => undefined,
): Promise<void> {
// ManagementStoreLogic is the in-DO synchronous implementation of every
// ManagementStore operation; the shared drain awaits its return values, so
// one implementation owns claim, backoff, terminal settling, and logging.
const local = localSlackPresentationState(stores, env);
// A runner-executed turn's presentation lives in its thread runner.
const runnerCopy = (runId: string) => {
const runner = presentationRunner(runId);
return runner
? new CfSlackStateStore({
slackPresentationGet: (id: string) => runner.presentationGet(id),
slackPresentationTransition: (input: SlackPresentationTransitionInput) =>
runner.presentationTransition(input),
} as unknown as TagStateRpc)
: undefined;
};
const presentation = {
state: {
...local,
getRunPresentation: (runId: string) =>
(runnerCopy(runId) ?? local).getRunPresentation!(runId),
transitionRunPresentation: (input: SlackPresentationTransitionInput) =>
(runnerCopy(input.runId) ?? local).transitionRunPresentation!(input),
} satisfies SlackPresentationStatePort,
resolveClient: async (workspaceId: string) =>
(await resolveInstallation(workspaceId)).client,
};
const work = stores.work as unknown as WorkStore;
await drainManagementReceiptOutbox({
management: stores.management as unknown as ManagementStore,
onDeliveredSettled: (record) => completeSettledAgentWelcomeHandoff(
record,
stores.config,
stores.management as unknown as ManagementStore,
),
onTerminalFailure: (record) => failAgentWelcomeTurn(
record,
presentation,
(turnJobId) => stores.turnJobs.markError(turnJobId),
work,
),
// Wakes admitted here go into this state store's own queue, as its asks do.
wake: (request) => admitSetupWake(request, env, localHostAddressedAdmission(stores, env)),
deliver: (record) => deliverManagementReceiptToSlack(record, {
identity: stores.identity as unknown as IdentityStore,
resolveInstallation,
onDelivered: async (deliveredRecord, delivery) => {
try {
await completeAgentWelcomeDelivery(
deliveredRecord,
delivery,
stores.config,
presentation,
work,
);
} finally {
if (isAgentCreatedWelcome(deliveredRecord.receipt) &&
deliveredRecord.receipt.turnJobId) {
stores.slack.endTurn(
deliveredRecord.receipt.turnJobId, stores.turnJobs, stores.work, stores.presentations,
);
}
}
},
}),
});
}
async function drainCloudflareScheduleActions(
stores: TagStateStores,
platformEnv: PlatformEnv,
): Promise<{ attempted: number; nextDueAt?: number }> {
const now = Date.now();
const nextDueAt = stores.routines.nextScheduleActionDueAt();
if (nextDueAt === undefined || nextDueAt > now) {
const local = localGatewayAppStores(stores);
await reconcileScheduleActionReceipts({
routines: local.routines,
management: local.management,
at: now,
});
return { attempted: 0, ...(nextDueAt !== undefined ? { nextDueAt } : {}) };
}
const { local, service } = localManagementRuntime(stores, platformEnv);
return retryDueSlackScheduleActions({
dependencies: {
management: local.management,
routines: local.routines,
service,
owner: `alarm:schedule:${Date.now()}`,
},
resolveContext: async (action, request) => {
return {
userId: action.actorUserId,
membershipId: action.actorMembershipId,
organizationId: request.organizationId,
actingAgentId: action.agentId,
origin: {
kind: 'slack',
workspaceId: action.workspaceId,
channelId: action.channelId,
threadTs: action.threadTs,
messageTs: action.messageTs,
conversationKind: action.conversationKind,
agentId: action.agentId,
},
};
},
});
}
async function drainLedgerRuns(
stores: TagStateStores,
platformEnv: PlatformEnv,
resolveInstallation: SlackInstallationExecutionResolver,
productTelemetry: ProductTelemetryCapture,
): Promise<RunDriverDrainResult> {
return new DurableRunDriver(stores.work, {
ownerId: 'cloudflare_ledger_run_driver',
authorityEpoch: 1,
leaseDurationMs: 30_000,
maxClaims: 4,
concurrency: 4,
handle: createLedgerSlackRunHandler({
// WorkStoreLogic is the in-DO synchronous implementation of every
// WorkStore operation; awaiting its return values preserves the same
// handler contract without a self-RPC through CfWorkStore.
work: stores.work as unknown as WorkStore,
turns: stores.turnJobs,
resolveInstallation,
verifyInstallationAccess: verifySlackInstallationTurnAccess,
platformEnv,
settingsStore: localSettingsStore(stores),
usageStore: localUsageStore(stores),
presentationState: localSlackPresentationState(stores, platformEnv),
setActiveWork: (key, generation, active) =>
stores.slack.setActiveWork(key, generation, active),
markCodingActiveWork: (key, generation) =>
stores.slack.markCodingActiveWork(key, generation),
onPublicMessageDelivered: (turn, assignment, delivery) =>
recordDeliveredSlackAgentMessage(stores.config, turn, assignment, delivery),
productTelemetry,
}),
}).drain();
}
async function drainGatewayInbox(
stores: TagStateStores,
platformEnv: PlatformEnv,
/**
* Called after each admitted delivery (runner mode: hand its turn over at
* once). With it, different conversations are admitted side by side.
*/
onAdmitted?: () => void,
): Promise<boolean> {
const pending = stores.gatewayInbox.claimPending(GATEWAY_INBOX_MAX_DRAIN_BATCH);
if (pending.length === 0) return stores.gatewayInbox.hasPending();
const appStores = localGatewayAppStores(stores);
let client: GatewayDeploymentClient;
try {
client = localGatewayDeploymentClient(appStores, platformEnv, createPlatformProductTelemetry({
env: platformEnv,
settings: appStores.settings,
config: appStores.config,
}));
} catch {
for (const item of pending) {
stores.gatewayInbox.retryOrRecover(item.id, 'delivery_dependency_unavailable');
}
return stores.gatewayInbox.hasPending();
}
// Runner mode: verify the bot identity while the deliveries are admitted,
// so each turn's hand-off can carry it (admitToRunner). A verification
// still fresh in this isolate is reused, and a failure here fails nothing.
const event = onAdmitted
? pending.map((item) => item.delivery).find((delivery) => delivery.kind === 'event.deliver')
: undefined;
const verifyingIdentity = event?.kind === 'event.deliver'
? resolveSlackInstallationExecutionContext(event.envelope.workspaceId, platformEnv, {
config: { getWorkspaceInstallation: (id) => stores.config.getWorkspaceInstallation(id) },
settings: appStores.settings,
gatewayClient: client,
}).then(() => undefined, () => undefined)
: undefined;
let needsRetry = false;
const uiExecution = { stores: appStores, enqueueTurn: localTurnEnqueue(stores) };
const admit = async (item: (typeof pending)[number]) => {
// A turn admitted from this delivery measures its latency from receipt:
// the delivery may have waited here while an earlier alarm ran turns.
const releaseReceipt = item.delivery.kind === 'event.deliver'
? stores.turnJobs.noteReceipt(item.delivery.envelope.eventId, item.acceptedAt)
: undefined;
try {
const outcome = item.delivery.kind === 'event.deliver'
? await processGatewaySlackEnvelope(
item.delivery.envelope,
platformEnv,
client,
uiExecution,
)
: item.delivery.kind === 'interaction.agent_selected'
? await processGatewayAgentSelection(
item.delivery,
platformEnv,
client,
appStores,
)
: item.delivery.kind === 'interaction.ui_action'
? await processGatewayUiAction(item.delivery, platformEnv, client, uiExecution)
: item.delivery.kind === 'interaction.view_submission'
? await processGatewayViewSubmission(item.delivery, platformEnv, client, uiExecution)
: await processGatewayPrivateChannelSetup(
item.delivery,
platformEnv,
client,
appStores,
);
if (outcome === 'accepted') {
stores.gatewayInbox.complete(item.id);
onAdmitted?.();
} else {
stores.gatewayInbox.markRecoveryRequired(item.id, 'binding_revalidation_rejected');
}
} catch (error) {
const retryDelayMs = gatewayDeliveryRetryDelayMs(item.attempts, error);
const retry = stores.gatewayInbox.retryOrRecover(
item.id,
gatewayDeliveryFailureReason(error),
retryDelayMs,
);
if (retry === 'recovery_required') {
recordGatewayDeliveryDeadLetter(item.delivery, item.attempts, error);
}
// A row in backoff is not due yet: the alarm arms for its due time
// (nextPendingDueAt) instead of re-polling every few seconds.
needsRetry ||= retry === 'pending' && retryDelayMs === 0;
} finally {
releaseReceipt?.();
}
};
// Deliveries of one conversation keep their order; different conversations
// are admitted side by side, so one slow delivery never holds another
// thread's first status.
const groups = new Map<string, Array<(typeof pending)[number]>>();
for (const item of pending) {
const key = gatewayConversationKey(item.delivery);
groups.set(key, [...(groups.get(key) ?? []), item]);
}
const queue = [...groups.values()];
await Promise.all(Array.from(
{ length: Math.min(onAdmitted ? GATEWAY_INBOX_CONVERSATION_CONCURRENCY : 1, queue.length) },
async () => {
for (let group = queue.shift(); group; group = queue.shift()) {
for (const item of group) await admit(item);
}
},
));
await verifyingIdentity;
return needsRetry || stores.gatewayInbox.hasPending();
}
/** Conversations admitted side by side by one inbox drain. */
const GATEWAY_INBOX_CONVERSATION_CONCURRENCY = 4;
/**
* The Slack conversation a delivery belongs to: its channel and thread root.
* Anything without one (installation, profile and interaction events) shares
* one ordered group.
*/
function gatewayConversationKey(delivery: GatewayInboundDelivery): string {
return delivery.kind === 'event.deliver' ? slackEventConversationKey(delivery.envelope.event) : 'other';
}
/** A Slack event's channel and thread root; `other` when it names none. */
function slackEventConversationKey(event: unknown): string {
const { channel, thread_ts: threadTs, ts } = event as { channel?: unknown; thread_ts?: unknown; ts?: unknown };
const root = typeof threadTs === 'string' ? threadTs : ts;
return typeof channel === 'string' && typeof root === 'string' ? `${channel}:${root}` : 'other';
}
/**
* Asks made by a turn this state store runs itself (the legacy alarm
* executor) are admitted here, against the local stores and queue, the way
* the gateway inbox admits a Slack event.
*/
function localAgentAskDispatcher(
stores: TagStateStores,
platformEnv: PlatformEnv,
productTelemetry: ProductTelemetryCapture,
): SlackAgentAskDispatcher {
return (request) => processSlackAgentAsks(request, platformEnv, localHostAddressedAdmission(stores, platformEnv, productTelemetry));
}
/** A front-desk hand-off made by a turn this state store runs itself, admitted like its asks. */
function localFrontDeskDispatcher(
stores: TagStateStores,
platformEnv: PlatformEnv,
productTelemetry: ProductTelemetryCapture,
): SlackFrontDeskHandoffDispatcher {
return (request) =>
processSlackFrontDeskHandoff(request, platformEnv, localHostAddressedAdmission(stores, platformEnv, productTelemetry));
}
/**
* What a host-addressed turn (an ask, a front-desk hand-off, or a setup's
* wake) admitted inside this state store runs on: its own stores, queue and
* gateway client.
*/
function localHostAddressedAdmission(
stores: TagStateStores,
platformEnv: PlatformEnv,
productTelemetry?: ProductTelemetryCapture,
) {
const appStores = localGatewayAppStores(stores);
let gatewayClient: GatewayDeploymentClient | undefined;
try {
gatewayClient = localGatewayDeploymentClient(appStores, platformEnv, productTelemetry ?? createPlatformProductTelemetry({
env: platformEnv,
settings: appStores.settings,
config: appStores.config,
}));
} catch {
// A direct installation needs no gateway client.
}
return {
stores: appStores,
...(gatewayClient ? { gatewayClient } : {}),
enqueueTurn: localTurnEnqueue(stores),
};
}
/** The gateway client of a turn admitted inside this state store, on its own stores. */
function localGatewayDeploymentClient(
appStores: AppStores,
platformEnv: PlatformEnv,
productTelemetry: ProductTelemetryCapture,
): GatewayDeploymentClient {
return new GatewayDeploymentClient({
settings: appStores.settings,
config: appStores.config,
identity: appStores.identity,
keyring: loadCredentialKeyring(platformEnv),
gatewayBaseUrl: resolveChickpeaGatewayUrl(platformEnv),
productTelemetry,
});
}
async function loadKeyFormFor(delivery: GatewayInboundDelivery): Promise<typeof import('./slack/key-form.ts') | undefined> {
return isKeyFormDelivery(delivery) ? import('./slack/key-form.ts') : undefined;
}
/**
* This state store's settings for the work it runs itself: management, the
* alarm's turns and ledger runs. An installation of a deployment serving many
* reads and writes its saved model keys only through a store's model
* credential methods, so this port carries them.
*/
function localSettingsStore(stores: TagStateStores): SettingsStore & ModelCredentialStore {
return {
getSetting: async (key) => stores.settings.getSetting(key),
getSettings: async (keys) => stores.settings.getSettings(keys),
setSetting: async (key, value) => stores.settings.setSetting(key, value),
deleteSetting: async (key) => stores.settings.deleteSetting(key),
applySettingsPatch: async (patch) => stores.settings.applySettingsPatch(patch),
mergeSettingStringSet: async (key, values) =>
stores.settings.mergeSettingStringSet(key, values),
readModelCredential: async (providerId) => stores.settings.readModelCredential(providerId),
publishModelCredential: async (input) => stores.settings.publishModelCredential(input),
rewrapModelCredential: async (input) => stores.settings.rewrapModelCredential(input),
};
}
function localManagementRuntime(
stores: TagStateStores,
platformEnv: PlatformEnv,
local: AppStores = localGatewayAppStores(stores),
) {
const settings = localSettingsStore(stores);
return {
local,
service: createLiveWorkspaceManagementService(platformEnv, {
identity: local.identity,
settings,
usage: local.usage,
slackCredentials: {
state: local.identity,
keyring: loadCredentialKeyring(platformEnv),
},
overrides: {
identity: local.identity,
config: local.config,
management: local.management,
memory: local.memory,
routines: local.routines,
work: local.work,
typedSlackMessage: async (query) => stores.turnJobs.typedSlackMessage(query),
setupBaseUrl: () => resolveSlackPublicUrl(platformEnv, settings),
},
}),
};
}
/**
* The alarm executor's turn-row port: the state store's own rows, except
* that a delivery and the stopped ending also settle the Runs and
* presentations of the turns that ended without running, as the runner's
* `markDelivered` and slackTurnStopFinish do. Spelled out like
* runnerTurnJobsPort, so the methods that differ are visible.
*/
function alarmTurnJobsPort(stores: TagStateStores): TurnExecutionPorts['turnJobs'] {
const rows = stores.turnJobs;
return {
recordAttempt: (...args) => rows.recordAttempt(...args),
markRecoveryRequired: (...args) => rows.markRecoveryRequired(...args),
prepareFlueDispatch: (...args) => rows.prepareFlueDispatch(...args),
reconcileFlueExistingInstance: (...args) => rows.reconcileFlueExistingInstance(...args),
recordFlueReceipt: (...args) => rows.recordFlueReceipt(...args),
recordFlueSettlement: (...args) => rows.recordFlueSettlement(...args),
recordPullRequest: (...args) => rows.recordPullRequest(...args),
freezeRuntimePlan: (...args) => rows.freezeRuntimePlan(...args),
getBoundRuntimePlan: (...args) => rows.getBoundRuntimePlan(...args),
getThreadContinuation: (...args) => rows.getThreadContinuation(...args),
recordUsagePersistence: (...args) => rows.recordUsagePersistence(...args),
recordInteractionIntent: (...args) => rows.recordInteractionIntent(...args),
recordSlackInteractionProgress: (...args) => rows.recordSlackInteractionProgress(...args),
markDelivered: (id) => stores.slack.endTurn(id, rows, stores.work, stores.presentations),
markError: (...args) => rows.markError(...args),
finishStop: (headId, outcome) =>
stores.slack.finishTurnStop(headId, outcome, rows, stores.work, stores.presentations),
};
}
function localSlackPresentationState(stores: TagStateStores, env: PlatformEnv): SlackPresentationStatePort {
return localSlackPresentationStatePort({
presentations: stores.presentations,
sharedReadsEnv: env,
matchFlueObservation: (instanceId, submissionId) =>
stores.turnJobs.matchFlueObservation(instanceId, submissionId),
});
}
function localUsageStore(stores: TagStateStores): UsageStore {
return {
admitOperation: async (input) => stores.usage.admitOperation(input),
recordTerminal: async (input) => stores.usage.recordTerminal(input),
recordConnectorUsage: async (input) => stores.usage.recordConnectorUsage(input),
recordModelRequest: async (record) => stores.usage.recordModelRequest(record),
getModelRequest: async (requestId) => stores.usage.getModelRequest(requestId),
reserveConnectorQuota: async (input) => stores.usage.reserveConnectorQuota(input),
releaseConnectorQuota: async (input) => stores.usage.releaseConnectorQuota(input),
summarizeConnectorUsage: async (query) => stores.usage.summarizeConnectorUsage(query),
getOperation: async (operationId) => stores.usage.getOperation(operationId),
getOperationByRunId: async (runId) => stores.usage.getOperationByRunId(runId),
listOperations: async (query) => stores.usage.listOperations(query),
summarize: async (query) => stores.usage.summarize(query),
listActivityAttributions: async (query) => stores.usage.listActivityAttributions(query),
putCredential: async (input) => stores.usage.putCredential(input),
retireCredential: async (credentialRefId, version, retiredAt) =>
stores.usage.retireCredential(credentialRefId, version, retiredAt),
listCredentials: async (providerId) => stores.usage.listCredentials(providerId),
cleanupRetention: async (at) => stores.usage.cleanupRetention(at),
getRetentionStatus: async () => stores.usage.getRetentionStatus(),
listUsageAuditEvents: async (limit) => stores.usage.listUsageAuditEvents(limit),
};
}
function earliestDefined(...values: Array<number | undefined>): number | undefined {
const defined = values.filter((value): value is number => value !== undefined);
return defined.length ? Math.min(...defined) : undefined;
}
function workspaceManagementRpcFailure(): WorkspaceManagementToolResult {
return {
ok: false,
error: {
code: 'management_error',
message: 'The workspace management request failed.',
},
};
}
function rpcError(
code: StateRpcErrorCode,
message: string,
details?: Record<string, string>,
): { ok: false; error: { code: typeof code; message: string; details?: Record<string, string> } } {
return { ok: false, error: { code, message, ...(details ? { details } : {}) } };
}
export default createRoutineScheduledHandler({
heartbeat: runRoutineHeartbeat,
maintenance: runWorkMaintenance,
duties: [keepOAuthCredentialsAlive],
});
/** Its own duty, so renewals never delay maintenance or the gateway wake. */
async function keepOAuthCredentialsAlive(
scheduledTime: number,
rawEnv: Record<string, unknown>,
): Promise<void> {
if (!isOAuthKeepAliveMinute(scheduledTime)) return;
const platformEnv = rawEnv as PlatformEnv;
await runOAuthKeepAliveSweep({
config: getConfigStore(platformEnv),
settings: getSettingsStore(platformEnv),
});
}
async function runWorkMaintenance(
scheduledTime: number,
rawEnv: Record<string, unknown>,
): Promise<void> {
await runWithGuaranteedFinalizer(async () => {
const result = await tagStateStub(rawEnv).maintainWork(scheduledTime);
if (!result.ok) {
throw new Error(`Work maintenance failed: ${result.error.message}`);
}
const platformEnv = rawEnv as PlatformEnv;
try { await purgeExpiredImageOutputs(getSettingsStore(platformEnv), scheduledTime); }
catch { console.warn('[chickpea] Image cache maintenance did not complete'); }
if (isContainerLeaseSweepMinute(scheduledTime)) {
try { await maintainWorkspaceContainerLeases(platformEnv, scheduledTime); }
catch { console.warn('[chickpea] Coding workspace container lease cleanup did not complete'); }
try { await maintainWebsiteLogins(platformEnv); }
catch { console.warn('[chickpea] Website login cleanup did not complete'); }
try { await maintainBrowserSessionLeases(platformEnv, scheduledTime); }
catch { console.warn('[chickpea] Browser session lease cleanup did not complete'); }
}
if (isCheckpointSweepMinute(scheduledTime)) {
try { await sweepInstallationCheckpoints(platformEnv, scheduledTime); }
catch { console.warn('[chickpea] Coding workspace checkpoint cleanup did not complete'); }
}
await repairPendingOAuthContinuationResumes({
settings: getSettingsStore(platformEnv),
onReady: async (continuation) => {
if (!(await isOAuthContinuationActorActive({
continuation,
identity: getIdentityStore(platformEnv),
}))) {
throw new Error('OAuth continuation member is no longer active.');
}
const resumed = await getSlackStateStore(platformEnv).resumeTurnAfterOAuth?.(
continuation.taskId,
continuation.id,
);
if (!resumed) throw new Error('OAuth continuation task is unavailable.');
},
});
}, async () => {
// The gateway session is the ingress lifeline for the shared Slack lane.
// Work or OAuth repair failures must never suppress its periodic wake.
await wakeCloudflareGatewaySession(rawEnv);
});
}
export { SlackGatewaySession };
// Per-thread turn executor (migration v11), the default for Cloudflare turns.
export { SlackThreadRunner } from './slack/thread-runner.ts';
async function runRoutineHeartbeat(
scheduledTime: number,
owner: string,
rawEnv: Record<string, unknown>,
context: { waitUntil(promise: Promise<unknown>): void },
): Promise<void> {
const productTelemetry = createPlatformProductTelemetry({
env: rawEnv,
settings: getSettingsStore(rawEnv),
config: getConfigStore(rawEnv),
lifecycle: createWaitUntilTelemetryLifecycle(context),
});
await runSharedRoutineHeartbeat({
scheduledTime,
owner,
env: rawEnv,
productTelemetry,
});
}