import { defineSchema, defineTable } from "convex/server"; import { v } from "convex/values"; export default defineSchema({ daemonDefinitions: defineTable({ daemonId: v.string(), name: v.string(), desiredState: v.union(v.literal("online"), v.literal("offline")), agentOsConfig: v.any(), configVersion: v.number(), createdAt: v.number(), updatedAt: v.number(), }).index("by_daemonId", ["daemonId"]), daemonPresence: defineTable({ daemonId: v.string(), sessionId: v.string(), status: v.union( v.literal("online"), v.literal("draining"), v.literal("offline") ), hostname: v.string(), platform: v.string(), architecture: v.string(), version: v.string(), startedAt: v.number(), lastHeartbeatAt: v.number(), }) .index("by_daemonId", ["daemonId"]) .index("by_sessionId", ["sessionId"]), daemonCommands: defineTable({ daemonId: v.string(), method: v.string(), args: v.any(), actorKey: v.array(v.string()), status: v.union( v.literal("queued"), v.literal("claimed"), v.literal("running"), v.literal("succeeded"), v.literal("failed"), v.literal("cancelled") ), attempts: v.number(), claimedBySessionId: v.optional(v.string()), leaseExpiresAt: v.optional(v.number()), createdAt: v.number(), updatedAt: v.number(), startedAt: v.optional(v.number()), completedAt: v.optional(v.number()), result: v.optional(v.any()), error: v.optional(v.string()), }) .index("by_daemonId_and_status", ["daemonId", "status"]) .index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]), daemonCommandEvents: defineTable({ daemonId: v.string(), commandId: v.optional(v.id("daemonCommands")), kind: v.string(), data: v.any(), createdAt: v.number(), }) .index("by_daemonId_and_createdAt", ["daemonId", "createdAt"]) .index("by_commandId_and_createdAt", ["commandId", "createdAt"]), projects: defineTable({ ownerId: v.string(), provider: v.literal("github"), providerRepoId: v.number(), name: v.string(), description: v.optional(v.string()), repoOwner: v.string(), repoName: v.string(), repoUrl: v.string(), defaultBranch: v.string(), createdAt: v.number(), updatedAt: v.number(), }) .index("by_owner_and_createdAt", ["ownerId", "createdAt"]) .index("by_owner_and_repository", [ "ownerId", "provider", "repoOwner", "repoName", ]), projectArtifacts: defineTable({ projectId: v.id("projects"), path: v.string(), content: v.string(), revision: v.number(), createdAt: v.number(), updatedAt: v.number(), }) .index("by_project", ["projectId"]) .index("by_project_and_path", ["projectId", "path"]), projectIssues: defineTable({ projectId: v.id("projects"), number: v.number(), title: v.string(), body: v.string(), status: v.union( v.literal("open"), v.literal("queued"), v.literal("working"), v.literal("needs-input"), v.literal("completed"), v.literal("failed") ), createdAt: v.number(), updatedAt: v.number(), }) .index("by_project_and_number", ["projectId", "number"]) .index("by_project_and_status", ["projectId", "status"]), projectWorkRuns: defineTable({ projectId: v.id("projects"), issueId: v.id("projectIssues"), agentId: v.string(), daemonId: v.string(), actorKey: v.array(v.string()), status: v.union( v.literal("ready"), v.literal("queued"), v.literal("working"), v.literal("needs-input"), v.literal("completed"), v.literal("failed") ), summary: v.optional(v.string()), createdAt: v.number(), updatedAt: v.number(), startedAt: v.optional(v.number()), completedAt: v.optional(v.number()), }) .index("by_issue", ["issueId"]) .index("by_project_and_createdAt", ["projectId", "createdAt"]), projectSignals: defineTable({ projectId: v.id("projects"), issueId: v.optional(v.id("projectIssues")), runId: v.optional(v.id("projectWorkRuns")), kind: v.string(), data: v.any(), createdAt: v.number(), }) .index("by_project_and_createdAt", ["projectId", "createdAt"]) .index("by_issue_and_createdAt", ["issueId", "createdAt"]), todos: defineTable({ text: v.string(), completed: v.boolean(), }), // ----------------------------------------------------------------- // Flue persistence stores (schema/format version 4). // ----------------------------------------------------------------- // One-row key/value metadata table, keyed with `schema_version` for the // persisted-store versioning obligation. flueMeta: defineTable({ key: v.string(), value: v.string(), }).index("by_key", ["key"]), // Durable agent-submission lifecycle: one row per submission. Settlement // obligations live on the same row so reserve/finalize are atomic with the // status transition they gate. flueSubmissions: defineTable({ submissionId: v.string(), sessionKey: v.string(), sequence: v.number(), kind: v.union(v.literal("dispatch"), v.literal("direct")), // Client-supplied canonical input/chunks encoded by Flue helpers; exact // string compared on idempotent replay. inputJson: v.string(), chunksJson: v.string(), traceCarrierJson: v.optional(v.string()), // Admit-time canonical record (Flue ConversationCreatedRecord) for the // default session of the default harness. recordJson: v.optional(v.string()), status: v.union( v.literal("queued"), v.literal("running"), v.literal("terminalizing"), v.literal("settled") ), acceptedAt: v.number(), canonicalReadyAt: v.optional(v.number()), attemptId: v.optional(v.string()), inputAppliedAt: v.optional(v.number()), recoveryRequestedAt: v.optional(v.number()), abortRequestedAt: v.optional(v.number()), startedAt: v.optional(v.number()), error: v.optional(v.string()), attemptCount: v.number(), maxRetry: v.number(), timeoutAt: v.number(), ownerId: v.optional(v.string()), leaseExpiresAt: v.number(), // Settlement obligation (reserved/finalized on this row). settlementRecordId: v.optional(v.string()), settlementRecordJson: v.optional(v.string()), settledOutcome: v.optional( v.union(v.literal("completed"), v.literal("failed"), v.literal("aborted")) ), updatedAt: v.number(), }) .index("by_submissionId", ["submissionId"]) .index("by_sessionKey_and_sequence", ["sessionKey", "sequence"]) .index("by_status_and_sequence", ["status", "sequence"]) .index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]) .index("by_status_and_settlementRecordId", [ "status", "settlementRecordId", ]), // Durable evidence that a submission attempt started and has not yet // settled. Append-once by (submissionId, attemptId). flueAttemptMarkers: defineTable({ submissionId: v.string(), attemptId: v.string(), createdAt: v.number(), }) .index("by_submissionId_and_attemptId", ["submissionId", "attemptId"]) .index("by_submissionId", ["submissionId"]), // Conversation-stream metadata: one row per stream path. flueConversationStreams: defineTable({ path: v.string(), identityJson: v.string(), incarnation: v.string(), producerId: v.optional(v.string()), producerEpoch: v.number(), nextProducerSequence: v.number(), nextOffset: v.number(), closed: v.boolean(), createdAt: v.number(), }).index("by_path", ["path"]), // Conversation-stream batches: one row per appended batch. Offset is // 0-based; `seq` is the row's position in the stream. flueConversationBatches: defineTable({ path: v.string(), seq: v.number(), producerId: v.string(), producerEpoch: v.number(), producerSequence: v.number(), recordsJson: v.string(), submissionId: v.optional(v.string()), attemptId: v.optional(v.string()), appendedAt: v.number(), }) .index("by_path_and_seq", ["path", "seq"]) .index("by_path_producer_epoch_producerSequence", [ "path", "producerId", "producerEpoch", "producerSequence", ]), // Event-stream metadata: one row per stream path. flueEventStreams: defineTable({ path: v.string(), nextSeq: v.number(), closed: v.boolean(), createdAt: v.number(), }).index("by_path", ["path"]), // Event-stream entries: one row per appended event. `onceKey` carries the // idempotency key for `appendEventOnce` and is null for plain appends. flueEventEntries: defineTable({ path: v.string(), seq: v.number(), onceKey: v.optional(v.string()), dataJson: v.string(), appendedAt: v.number(), }) .index("by_path_and_seq", ["path", "seq"]) .index("by_path_and_onceKey", ["path", "onceKey"]), // Workflow run records. flueRuns: defineTable({ runId: v.string(), workflowName: v.string(), status: v.union( v.literal("active"), v.literal("completed"), v.literal("errored") ), startedAt: v.string(), inputJson: v.optional(v.string()), traceCarrierJson: v.optional(v.string()), endedAt: v.optional(v.string()), isError: v.optional(v.boolean()), durationMs: v.optional(v.number()), resultJson: v.optional(v.string()), errorJson: v.optional(v.string()), }) .index("by_runId", ["runId"]) .index("by_startedAt_and_runId", ["startedAt", "runId"]) .index("by_workflowName_and_startedAt_and_runId", [ "workflowName", "startedAt", "runId", ]) .index("by_status_startedAt_runId", ["status", "startedAt", "runId"]) .index("by_status_workflowName_startedAt_runId", [ "status", "workflowName", "startedAt", "runId", ]), // Immutable attachment bytes. Identity is (streamPath, attachmentId); reads // are additionally scoped by conversationId. flueAttachments: defineTable({ streamPath: v.string(), conversationId: v.string(), attachmentId: v.string(), mimeType: v.string(), size: v.number(), digest: v.string(), filename: v.optional(v.string()), bytes: v.bytes(), createdAt: v.number(), }) .index("by_streamPath", ["streamPath"]) .index("by_streamPath_and_attachmentId", ["streamPath", "attachmentId"]) .index("by_streamPath_and_conversationId_and_attachmentId", [ "streamPath", "conversationId", "attachmentId", ]), });