Add Flue persistence backend on Convex

- Define the schema v4 Flue tables (submissions, attempt markers,
  conversation streams/batches, event streams/entries, runs, attachments,
  meta) in schema.ts with the indexes the adapter relies on.
- Implement fluePersistence.ts: token-gated queries/mutations covering
  submission lifecycle, leasing, settlement, conversation/event streams,
  runs, and attachments.

packages/backend/convex/schema.ts

@@ -66,4 +66,188 @@ export default defineSchema({
     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",
+    ])
 });