diff --git a/packages/backend/convex/README.md b/packages/backend/convex/README.md index d16fc64..2d035d9 100644 --- a/packages/backend/convex/README.md +++ b/packages/backend/convex/README.md @@ -1,17 +1,11 @@ # Convex Backend -Convex is the only application API used by web, desktop, and mobile clients. The active backend covers the durable control plane: tenancy, projects, conversation turns, Signals, Work planning (Definition/Design without approval gates), real Runs and Attempts through AgentOS, artifacts/delivery metadata, and Flue persistence. +Convex is the only application API used by web clients. The active backend covers the durable tenancy and Git onboarding control plane: organizations, Git provider connections, repository discovery/sync, public-Git import, and projects with repository context. ## Domain relations - `organizations` and `organizationMembers` own tenancy. - `projects` owns the single v0 Git source; `projectContextDocuments` stores repository context. -- `conversations`, `conversationTurns`, `conversationMessages`, and `conversationAttachments` own chat admission and reactive history. -- `signals`, `signalConstraints`, and `signalSources` preserve structured intent and exact evidence. -- `works`, `signalWorkAttachments`, and `workEvents` own proposed Work and provenance. -- `workDefinitions` and `workQuestions` own versioned outcome contracts and structured questions. -- `designPackets` and `workSlices` preserve versioned implementation intent and executable slice order. -- `workRuns`, `workAttempts`, `workAttemptEvents`, and `resolverDecisions` own bounded real execution and recovery. -- `workArtifacts` and `workDeliveries` preserve evidence and delivery metadata without performing Git or sandbox work. +- `gitProviderAccounts`, `gitConnections`, `gitProviderOrganizations`, `gitRepositories`, `gitMigrations`, and `gitWebhookDeliveries` own provider onboarding, repo sync, migration, and webhook delivery state. -The `flue*` tables are infrastructure tables required by Flue's persistence contract. They are deliberately separate from the product relations. +Retired agent, conversation, Signal, Work, and Flue persistence tables were deleted from the schema as part of the 2026-08-01 tenancy cutover; `cleanup.ts` holds the bounded-batch deleter used to retire their data. diff --git a/packages/backend/convex/_generated/api.d.ts b/packages/backend/convex/_generated/api.d.ts index 1d3d1b9..1a8357e 100644 --- a/packages/backend/convex/_generated/api.d.ts +++ b/packages/backend/convex/_generated/api.d.ts @@ -8,13 +8,10 @@ * @module */ -import type * as agentBackend from "../agentBackend.js"; import type * as auth from "../auth.js"; import type * as authz from "../authz.js"; -import type * as conversationMessages from "../conversationMessages.js"; -import type * as conversationProjections from "../conversationProjections.js"; +import type * as cleanup from "../cleanup.js"; import type * as crons from "../crons.js"; -import type * as fluePersistence from "../fluePersistence.js"; import type * as gitConnectionData from "../gitConnectionData.js"; import type * as gitConnectionHealth from "../gitConnectionHealth.js"; import type * as gitConnections from "../gitConnections.js"; @@ -26,11 +23,6 @@ import type * as organizations from "../organizations.js"; import type * as privateData from "../privateData.js"; import type * as projects from "../projects.js"; import type * as publicGit from "../publicGit.js"; -import type * as signalRouting from "../signalRouting.js"; -import type * as workArtifacts from "../workArtifacts.js"; -import type * as workExecution from "../workExecution.js"; -import type * as workPlanning from "../workPlanning.js"; -import type * as works from "../works.js"; import type { ApiFromModules, @@ -39,13 +31,10 @@ import type { } from "convex/server"; declare const fullApi: ApiFromModules<{ - agentBackend: typeof agentBackend; auth: typeof auth; authz: typeof authz; - conversationMessages: typeof conversationMessages; - conversationProjections: typeof conversationProjections; + cleanup: typeof cleanup; crons: typeof crons; - fluePersistence: typeof fluePersistence; gitConnectionData: typeof gitConnectionData; gitConnectionHealth: typeof gitConnectionHealth; gitConnections: typeof gitConnections; @@ -57,11 +46,6 @@ declare const fullApi: ApiFromModules<{ privateData: typeof privateData; projects: typeof projects; publicGit: typeof publicGit; - signalRouting: typeof signalRouting; - workArtifacts: typeof workArtifacts; - workExecution: typeof workExecution; - workPlanning: typeof workPlanning; - works: typeof works; }>; /** @@ -92,5 +76,4 @@ export declare const internal: FilterApi< export declare const components: { betterAuth: import("@convex-dev/better-auth/_generated/component.js").ComponentApi<"betterAuth">; - workflow: import("@convex-dev/workflow/_generated/component.js").ComponentApi<"workflow">; }; diff --git a/packages/backend/convex/_generated/server.d.ts b/packages/backend/convex/_generated/server.d.ts index 1ae0398..bc6ac21 100644 --- a/packages/backend/convex/_generated/server.d.ts +++ b/packages/backend/convex/_generated/server.d.ts @@ -25,9 +25,6 @@ import type { DataModel } from "./dataModel.js"; * Typesafe environment variables declared in `convex.config.ts`. */ type Env = { - readonly AGENT_BACKEND_URL: string | undefined; - readonly FLUE_DB_TOKEN: string; - readonly FLUE_URL: string | undefined; readonly GITEA_TOKEN: string | undefined; readonly GITEA_URL: string | undefined; readonly GITEA_WEBHOOK_SECRET: string | undefined; diff --git a/packages/backend/convex/artifacts.ts b/packages/backend/convex/artifacts.ts new file mode 100644 index 0000000..dbf2518 --- /dev/null +++ b/packages/backend/convex/artifacts.ts @@ -0,0 +1,27 @@ +import { v } from "convex/values"; + +import type { Doc } from "./_generated/dataModel"; +import { query } from "./_generated/server"; +import { requireArtifactMember, requireWorkMember } from "./authz"; + +export const get = query({ + args: { artifactId: v.id("artifacts") }, + handler: async (ctx, args): Promise | null> => { + await requireArtifactMember(ctx, args.artifactId); + return await ctx.db.get(args.artifactId); + }, +}); + +export const listForWork = query({ + args: { limit: v.optional(v.number()), workId: v.id("works") }, + handler: async (ctx, args): Promise[]> => { + const { organizationId } = await requireWorkMember(ctx, args.workId); + const limit = Math.min(Math.max(Math.floor(args.limit ?? 50), 1), 100); + return await ctx.db + .query("artifacts") + .withIndex("by_workId_and_createdAt", (q) => q.eq("workId", args.workId)) + .filter((q) => q.eq(q.field("organizationId"), organizationId)) + .order("desc") + .take(limit); + }, +}); diff --git a/packages/backend/convex/authz.ts b/packages/backend/convex/authz.ts index b4f814a..04548f7 100644 --- a/packages/backend/convex/authz.ts +++ b/packages/backend/convex/authz.ts @@ -79,6 +79,41 @@ export const requireProjectMember = async ( return { organizationId: project.organizationId, userId }; }; +export const requireWorkMember = async ( + ctx: QueryCtx | MutationCtx, + workId: Id<"works"> +): Promise<{ organizationId: Id<"organizations">; userId: string }> => { + const work = await ctx.db.get(workId); + if (!work) { + throw new ConvexError("Work not found"); + } + const userId = await requireOrganizationMember(ctx, work.organizationId); + return { organizationId: work.organizationId, userId }; +}; + +export const requireThreadMember = async ( + ctx: QueryCtx | MutationCtx, + threadId: Id<"threads"> +): Promise<{ organizationId: Id<"organizations">; userId: string }> => { + const thread = await ctx.db.get(threadId); + if (!thread) { + throw new ConvexError("Thread not found"); + } + return await requireWorkMember(ctx, thread.workId); +}; + +export const requireArtifactMember = async ( + ctx: QueryCtx | MutationCtx, + artifactId: Id<"artifacts"> +): Promise<{ organizationId: Id<"organizations">; userId: string }> => { + const artifact = await ctx.db.get(artifactId); + if (!artifact) { + throw new ConvexError("Artifact not found"); + } + const userId = await requireOrganizationMember(ctx, artifact.organizationId); + return { organizationId: artifact.organizationId, userId }; +}; + /** * Prove the authenticated user is an owner of the given organization. * Provisioning functions require owner-level authorization. diff --git a/packages/backend/convex/cleanup.ts b/packages/backend/convex/cleanup.ts new file mode 100644 index 0000000..0840511 --- /dev/null +++ b/packages/backend/convex/cleanup.ts @@ -0,0 +1,58 @@ +import { v } from "convex/values"; + +import type { TableNames } from "./_generated/dataModel"; +import { internalMutation } from "./_generated/server"; + +/** + * Temporary bounded-batch deleter for the 2026-08-01 tenancy cutover. + * Delete tables in RETIRED_TABLE_DELETE_ORDER, calling + * `deleteRetiredTableBatch` repeatedly with a small limit until a table + * returns fewer rows than the limit, then move to the next table. + */ +export const RETIRED_TABLE_DELETE_ORDER = [ + "flueAttachments", + "flueRuns", + "flueEventEntries", + "flueEventStreams", + "flueConversationBatches", + "flueConversationStreams", + "flueAttemptMarkers", + "flueSubmissions", + "flueMeta", + "workDeliveries", + "workArtifacts", + "resolverDecisions", + "workAttemptEvents", + "workAttempts", + "workRuns", + "workSlices", + "designPackets", + "workQuestions", + "workDefinitions", + "workEvents", + "signalWorkAttachments", + "works", + "signalSources", + "signalConstraints", + "signals", + "conversationAttachments", + "conversationMessages", + "conversationTurns", + "conversations", +] as const; + +export const deleteRetiredTableBatch = internalMutation({ + args: { + limit: v.number(), + table: v.string(), + }, + handler: async (ctx, args) => { + const batchSize = Math.min(Math.max(Math.floor(args.limit), 1), 500); + // The reduced schema no longer lists retired tables, so the name is cast. + const rows = await ctx.db.query(args.table as TableNames).take(batchSize); + for (const row of rows) { + await ctx.db.delete(row._id); + } + return { deleted: rows.length }; + }, +}); diff --git a/packages/backend/convex/convex.config.ts b/packages/backend/convex/convex.config.ts index 353fe07..8a828fc 100644 --- a/packages/backend/convex/convex.config.ts +++ b/packages/backend/convex/convex.config.ts @@ -1,13 +1,9 @@ import betterAuth from "@convex-dev/better-auth/convex.config"; -import workflow from "@convex-dev/workflow/convex.config.js"; import { defineApp } from "convex/server"; import { v } from "convex/values"; const app = defineApp({ env: { - AGENT_BACKEND_URL: v.optional(v.string()), - FLUE_DB_TOKEN: v.string(), - FLUE_URL: v.optional(v.string()), GITEA_TOKEN: v.optional(v.string()), GITEA_URL: v.optional(v.string()), GITEA_WEBHOOK_SECRET: v.optional(v.string()), @@ -21,6 +17,5 @@ const app = defineApp({ }, }); app.use(betterAuth); -app.use(workflow); export default app; diff --git a/packages/backend/convex/crons.ts b/packages/backend/convex/crons.ts index b9b240b..7c31c64 100644 --- a/packages/backend/convex/crons.ts +++ b/packages/backend/convex/crons.ts @@ -1,17 +1,5 @@ import { cronJobs, makeFunctionReference } from "convex/server"; -// Reconciles attempts whose worker lease expired (crash/restart). The target -// is referenced by string so this file does not depend on regenerated codegen. -const reconcileRef = makeFunctionReference< - "mutation", - Record, - { reconciled: number } | null ->("workExecution:reconcileExpiredAttempts"); -const reconcileConversationTurnsRef = makeFunctionReference< - "mutation", - Record, - { reconciled: number } ->("conversationMessages:reconcileExpiredTurns"); const reconcileGitConnectionsRef = makeFunctionReference< "action", Record, @@ -25,16 +13,6 @@ const backfillConnectionsRef = makeFunctionReference< const crons = cronJobs(); -crons.interval( - "reconcile expired work attempts", - { seconds: 30 }, - reconcileRef -); -crons.interval( - "reconcile expired conversation turns", - { seconds: 30 }, - reconcileConversationTurnsRef -); crons.cron( "reconcile stale git connections", "0 * * * *", diff --git a/packages/backend/convex/projectEnvironmentVariables.ts b/packages/backend/convex/projectEnvironmentVariables.ts new file mode 100644 index 0000000..6b605ae --- /dev/null +++ b/packages/backend/convex/projectEnvironmentVariables.ts @@ -0,0 +1,84 @@ +import { ConvexError, v } from "convex/values"; + +import type { Doc, Id } from "./_generated/dataModel"; +import { mutation, query } from "./_generated/server"; +import { requireProjectMember } from "./authz"; + +const category = v.union( + v.literal("required"), + v.literal("integration"), + v.literal("optional") +); +const status = v.union( + v.literal("missing"), + v.literal("configured"), + v.literal("invalid") +); +const source = v.union( + v.literal("detected"), + v.literal("user"), + v.literal("system") +); + +export const list = query({ + args: { projectId: v.id("projects") }, + handler: async ( + ctx, + args + ): Promise[]> => { + const { organizationId } = await requireProjectMember(ctx, args.projectId); + return await ctx.db + .query("projectEnvironmentVariables") + .withIndex("by_projectId", (q) => q.eq("projectId", args.projectId)) + .filter((q) => q.eq(q.field("organizationId"), organizationId)) + .order("asc") + .collect(); + }, +}); + +export const upsert = mutation({ + args: { + category, + description: v.optional(v.string()), + name: v.string(), + projectId: v.id("projects"), + secretReference: v.optional(v.string()), + source, + status, + }, + handler: async (ctx, args): Promise> => { + const { organizationId } = await requireProjectMember(ctx, args.projectId); + const name = args.name.trim(); + if (!name) { + throw new ConvexError("Environment variable name is required"); + } + const existing = await ctx.db + .query("projectEnvironmentVariables") + .withIndex("by_projectId_and_name", (q) => + q.eq("projectId", args.projectId).eq("name", name) + ) + .unique(); + const value = { + category: args.category, + description: args.description, + name, + organizationId, + projectId: args.projectId, + secretReference: args.secretReference, + source: args.source, + status: args.status, + updatedAt: Date.now(), + }; + const id: Id<"projectEnvironmentVariables"> = existing + ? existing._id + : await ctx.db.insert("projectEnvironmentVariables", value); + if (existing) { + await ctx.db.patch(id, value); + } + const saved = await ctx.db.get(id); + if (!saved) { + throw new ConvexError("Environment variable could not be saved"); + } + return saved; + }, +}); diff --git a/packages/backend/convex/projects.ts b/packages/backend/convex/projects.ts index 136bd28..aad05df 100644 --- a/packages/backend/convex/projects.ts +++ b/packages/backend/convex/projects.ts @@ -185,6 +185,101 @@ export const get = query({ }, }); +export const getSetup = query({ + args: { projectId: v.id("projects") }, + handler: async (ctx, args) => { + const { organizationId } = await requireProjectMember(ctx, args.projectId); + const events = await ctx.db + .query("events") + .withIndex("by_organizationId_and_projectId_and_recordedAt", (q) => + q.eq("organizationId", organizationId).eq("projectId", args.projectId) + ) + .order("desc") + .filter((q) => + q.or( + q.eq(q.field("type"), "project.setup.started"), + q.eq(q.field("type"), "project.setup.environment_required"), + q.eq(q.field("type"), "project.setup.environment_updated"), + q.eq(q.field("type"), "project.setup.ready"), + q.eq(q.field("type"), "project.setup.blocked"), + q.eq(q.field("type"), "project.preview.ready"), + q.eq(q.field("type"), "project.preview.blocked") + ) + ) + .take(50); + const artifacts = await ctx.db + .query("artifacts") + .withIndex("by_projectId_and_createdAt", (q) => + q.eq("projectId", args.projectId) + ) + .order("desc") + .filter((q) => + q.or( + q.eq(q.field("kind"), "project_setup"), + q.eq(q.field("kind"), "blocker"), + q.eq(q.field("kind"), "preview") + ) + ) + .take(20); + return { artifacts, events }; + }, +}); + +export const appendSetupEvent = internalMutation({ + args: { + artifactIds: v.optional(v.array(v.id("artifacts"))), + correlationId: v.string(), + idempotencyKey: v.string(), + payload: v.any(), + projectId: v.id("projects"), + type: v.union( + v.literal("project.setup.started"), + v.literal("project.setup.environment_required"), + v.literal("project.setup.environment_updated"), + v.literal("project.setup.ready"), + v.literal("project.setup.blocked"), + v.literal("project.preview.ready"), + v.literal("project.preview.blocked") + ), + visibility: v.union( + v.literal("timeline"), + v.literal("compact"), + v.literal("internal") + ), + }, + handler: async (ctx, args): Promise> => { + const project = await ctx.db.get(args.projectId); + if (!project) { + throw new ConvexError("Project not found"); + } + const existing = await ctx.db + .query("events") + .withIndex("by_organizationId_and_idempotencyKey", (q) => + q + .eq("organizationId", project.organizationId) + .eq("idempotencyKey", args.idempotencyKey) + ) + .unique(); + if (existing) { + return existing._id; + } + return await ctx.db.insert("events", { + actor: { kind: "system", service: "project-setup" }, + artifactIds: args.artifactIds, + correlationId: args.correlationId, + idempotencyKey: args.idempotencyKey, + occurredAt: Date.now(), + organizationId: project.organizationId, + payload: args.payload, + projectId: args.projectId, + recordedAt: Date.now(), + scopeKind: "global", + type: args.type, + visibility: args.visibility, + }); + }, +}); + export const importPublicGit = action({ args: { repositoryUrl: v.string() }, handler: async (ctx, args): Promise => { diff --git a/packages/backend/convex/schema.ts b/packages/backend/convex/schema.ts index 366ac95..5f8e1d5 100644 --- a/packages/backend/convex/schema.ts +++ b/packages/backend/convex/schema.ts @@ -3,32 +3,6 @@ import { v } from "convex/values"; /* eslint-disable sort-keys */ -const attemptClassification = v.union( - v.literal("Succeeded"), - v.literal("RetryableFailure"), - v.literal("NeedsInput"), - v.literal("Blocked"), - v.literal("VerificationFailed"), - v.literal("BudgetExhausted"), - v.literal("Cancelled"), - v.literal("PermanentFailure") -); -const workStatus = v.union( - v.literal("proposed"), - v.literal("defining"), - v.literal("awaiting-definition-approval"), - v.literal("designing"), - v.literal("awaiting-design-approval"), - v.literal("ready"), - v.literal("executing"), - v.literal("verifying"), - v.literal("needs-input"), - v.literal("blocked"), - v.literal("completed"), - v.literal("failed"), - v.literal("cancelled") -); - export default defineSchema({ organizations: defineTable({ createdAt: v.number(), @@ -237,561 +211,168 @@ export default defineSchema({ sourceUrl: v.string(), updatedAt: v.number(), }).index("by_projectId_and_path", ["projectId", "path"]), - conversations: defineTable({ - createdAt: v.number(), - organizationId: v.id("organizations"), - }).index("by_organizationId", ["organizationId"]), - conversationTurns: defineTable({ - attemptNumber: v.optional(v.number()), - clientRequestId: v.string(), - completedAt: v.optional(v.number()), - conversationId: v.id("conversations"), - createdAt: v.number(), - error: v.optional(v.string()), - leaseExpiresAt: v.optional(v.number()), - leaseOwner: v.optional(v.string()), - status: v.union( - v.literal("queued"), - v.literal("dispatching"), - v.literal("running"), - v.literal("completed"), - v.literal("failed"), - v.literal("aborted") + projectEnvironmentVariables: defineTable({ + category: v.union( + v.literal("required"), + v.literal("integration"), + v.literal("optional") ), - submissionId: v.optional(v.string()), - }) - .index("by_conversationId_and_clientRequestId", [ - "conversationId", - "clientRequestId", - ]) - .index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]) - .index("by_submissionId", ["submissionId"]), - conversationMessages: defineTable({ - content: v.string(), - conversationId: v.id("conversations"), - createdAt: v.number(), - ordinal: v.number(), - role: v.union(v.literal("user"), v.literal("assistant")), - turnId: v.id("conversationTurns"), - }) - .index("by_conversationId_and_ordinal", ["conversationId", "ordinal"]) - .index("by_turnId_and_role", ["turnId", "role"]), - conversationAttachments: defineTable({ - createdAt: v.number(), - filename: v.optional(v.string()), - messageId: v.id("conversationMessages"), - mimeType: v.string(), - storageId: v.id("_storage"), - }).index("by_messageId", ["messageId"]), - signals: defineTable({ - conversationId: v.id("conversations"), - createdAt: v.number(), - desiredOutcome: v.string(), + description: v.optional(v.string()), + name: v.string(), organizationId: v.id("organizations"), - processedByAgentInstanceId: v.string(), - processedByAgentName: v.string(), projectId: v.id("projects"), - sourceKey: v.string(), - summary: v.string(), - title: v.string(), + secretReference: v.optional(v.string()), + source: v.union( + v.literal("detected"), + v.literal("user"), + v.literal("system") + ), + status: v.union( + v.literal("missing"), + v.literal("configured"), + v.literal("invalid") + ), + updatedAt: v.number(), }) - .index("by_organization_and_createdAt", ["organizationId", "createdAt"]) - .index("by_organization_and_sourceKey", ["organizationId", "sourceKey"]) - .index("by_project_and_createdAt", ["projectId", "createdAt"]), - signalConstraints: defineTable({ - ordinal: v.number(), - signalId: v.id("signals"), - value: v.string(), - }).index("by_signalId_and_ordinal", ["signalId", "ordinal"]), - signalSources: defineTable({ - messageId: v.id("conversationMessages"), - ordinal: v.number(), - rawTextSnapshot: v.string(), - signalId: v.id("signals"), - sourceCreatedAt: v.number(), - }) - .index("by_signalId_and_ordinal", ["signalId", "ordinal"]) - .index("by_messageId", ["messageId"]), + .index("by_projectId", ["projectId"]) + .index("by_projectId_and_name", ["projectId", "name"]), works: defineTable({ createdAt: v.number(), - definitionApprovalVersion: v.optional(v.number()), - definitionVersion: v.optional(v.number()), - designVersion: v.optional(v.number()), - objective: v.string(), + createdBy: v.string(), + createdFromEventId: v.id("events"), organizationId: v.id("organizations"), - projectId: v.id("projects"), - status: workStatus, - title: v.string(), - updatedAt: v.number(), - }) - .index("by_project_and_createdAt", ["projectId", "createdAt"]) - .index("by_organization_and_createdAt", ["organizationId", "createdAt"]), - - signalWorkAttachments: defineTable({ - createdAt: v.number(), - signalId: v.id("signals"), - workId: v.id("works"), - }) - .index("by_signal", ["signalId"]) - .index("by_work", ["workId"]) - .index("by_signal_and_work", ["signalId", "workId"]), - - workEvents: defineTable({ - createdAt: v.number(), - idempotencyKey: v.string(), - kind: v.union( - v.literal("work.proposed"), - v.literal("signal.attached"), - v.literal("definition.requested"), - v.literal("definition.approved"), - v.literal("definition.saved"), - v.literal("definition.revised"), - v.literal("question.created"), - v.literal("question.answered"), - v.literal("question.withdrawn"), - v.literal("design.saved"), - v.literal("design.approved"), - v.literal("design.revised"), - v.literal("planner.failed"), - v.literal("slice.started"), - v.literal("slice.completed"), - v.literal("slice.ready"), - v.literal("run.started"), - v.literal("run.completed"), - v.literal("run.cancelled"), - v.literal("attempt.claimed"), - v.literal("attempt.event"), - v.literal("attempt.completed"), - v.literal("attempt.reconciled"), - v.literal("resolver.decided"), - v.literal("verification.started"), - v.literal("verification.completed"), - v.literal("verification.failed"), - v.literal("artifact.recorded"), - v.literal("delivery.recorded"), - v.literal("delivery.updated") - ), - payloadJson: v.optional(v.string()), - referenceId: v.optional(v.string()), - signalId: v.optional(v.id("signals")), - workId: v.id("works"), - }) - .index("by_work_and_createdAt", ["workId", "createdAt"]) - .index("by_work_and_idempotencyKey", ["workId", "idempotencyKey"]), - - workDefinitions: defineTable({ - createdAt: v.number(), - createdBy: v.string(), - payloadJson: v.string(), - risk: v.union(v.literal("low"), v.literal("medium"), v.literal("high")), - status: v.union( - v.literal("proposed"), - v.literal("current"), - v.literal("superseded") - ), - version: v.number(), - workId: v.id("works"), - }) - .index("by_work_and_version", ["workId", "version"]) - .index("by_work_and_status", ["workId", "status"]), - - workQuestions: defineTable({ - alternativesJson: v.string(), - answer: v.optional(v.string()), - createdAt: v.number(), - definitionVersion: v.number(), - impact: v.union(v.literal("low"), v.literal("medium"), v.literal("high")), - prompt: v.string(), - questionId: v.string(), - recommendation: v.optional(v.string()), - status: v.union( - v.literal("open"), - v.literal("answered"), - v.literal("withdrawn") - ), - workId: v.id("works"), - }) - .index("by_workId_and_definitionVersion", ["workId", "definitionVersion"]) - .index("by_workId_and_definitionVersion_and_questionId", [ - "workId", - "definitionVersion", - "questionId", - ]), - - designPackets: defineTable({ - createdAt: v.number(), - createdBy: v.string(), - definitionVersion: v.number(), - payloadJson: v.string(), - status: v.union( - v.literal("proposed"), - v.literal("current"), - v.literal("superseded") - ), - version: v.number(), - workId: v.id("works"), - }).index("by_work_and_version", ["workId", "version"]), - - workSlices: defineTable({ - createdAt: v.optional(v.number()), - designVersion: v.number(), - objective: v.string(), - observableBehavior: v.string(), - ordinal: v.number(), - payloadJson: v.string(), - sliceId: v.string(), - status: v.union( - v.literal("planned"), - v.literal("ready"), - v.literal("running"), - v.literal("completed"), - v.literal("blocked") - ), - title: v.string(), - workId: v.id("works"), - }) - .index("by_workId_and_designVersion", ["workId", "designVersion"]) - .index("by_workId_and_designVersion_and_sliceId", [ - "workId", - "designVersion", - "sliceId", - ]), - - workRuns: defineTable({ - baseRevision: v.optional(v.string()), - candidateRevision: v.optional(v.string()), - createdAt: v.number(), - designVersion: v.optional(v.number()), - endedAt: v.optional(v.number()), - kitId: v.string(), - kitVersion: v.string(), - scenario: v.optional(v.string()), - environmentId: v.optional(v.string()), - sliceId: v.optional(v.string()), - sliceRowId: v.optional(v.id("workSlices")), - startedAt: v.optional(v.number()), - status: v.union( - v.literal("ready"), - v.literal("running"), - v.literal("terminal"), - v.literal("cancelled") - ), - terminalClassification: v.optional(attemptClassification), - terminalSummary: v.optional(v.string()), - workflowId: v.optional(v.string()), - workId: v.id("works"), - }).index("by_work_and_createdAt", ["workId", "createdAt"]), - - workAttempts: defineTable({ - classification: v.optional(attemptClassification), - failureReason: v.optional( - v.union( - v.literal("Authentication"), - v.literal("Cancelled"), - v.literal("HarnessFailed"), - v.literal("InvalidInput"), - v.literal("ProviderUnavailable"), - v.literal("RepositoryFailed"), - v.literal("Timeout") - ) - ), - endedAt: v.optional(v.number()), - leaseExpiresAt: v.optional(v.number()), - leaseOwner: v.optional(v.string()), - number: v.number(), - runId: v.id("workRuns"), - startedAt: v.optional(v.number()), - status: v.union( - v.literal("queued"), - v.literal("claimed"), - v.literal("running"), - v.literal("terminal") - ), + primaryProjectId: v.optional(v.id("projects")), summary: v.optional(v.string()), - workId: v.id("works"), - workspaceKey: v.optional(v.string()), - }) - .index("by_runId_and_number", ["runId", "number"]) - .index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]), - - workAttemptEvents: defineTable({ - attemptId: v.id("workAttempts"), - kind: v.string(), - message: v.string(), - metadataJson: v.string(), - occurredAt: v.number(), - sequence: v.number(), - }).index("by_attempt_and_sequence", ["attemptId", "sequence"]), - - resolverDecisions: defineTable({ - attemptId: v.id("workAttempts"), - attemptNumber: v.number(), - classification: attemptClassification, - createdAt: v.number(), - decision: v.union(v.literal("retry"), v.literal("terminal")), - resultingWorkStatus: v.optional(workStatus), - runId: v.id("workRuns"), - summary: v.string(), - workId: v.id("works"), - }) - .index("by_runId_and_attemptNumber", ["runId", "attemptNumber"]) - .index("by_attemptId", ["attemptId"]), - - workArtifacts: defineTable({ - attemptId: v.optional(v.id("workAttempts")), - contentHash: v.optional(v.string()), - createdAt: v.number(), - designVersion: v.optional(v.number()), - environmentId: v.optional(v.string()), - idempotencyKey: v.string(), - kind: v.union( - v.literal("definition"), - v.literal("design"), - v.literal("diff"), - v.literal("test-report"), - v.literal("verification-report"), - v.literal("runtime-log"), - v.literal("screenshot"), - v.literal("video"), - v.literal("commit"), - v.literal("branch"), - v.literal("pull-request"), - v.literal("preview"), - v.literal("deployment"), - v.literal("other") - ), - metadataJson: v.string(), - organizationId: v.id("organizations"), - producer: v.string(), - projectId: v.id("projects"), - provenanceJson: v.string(), - runId: v.optional(v.id("workRuns")), - sliceId: v.optional(v.string()), - sourceRevision: v.optional(v.string()), title: v.string(), - uri: v.optional(v.string()), - verificationStatus: v.union( - v.literal("unverified"), - v.literal("verified"), - v.literal("rejected") - ), - workId: v.id("works"), }) - .index("by_workId_and_createdAt", ["workId", "createdAt"]) - .index("by_workId_and_idempotencyKey", ["workId", "idempotencyKey"]) - .index("by_runId_and_createdAt", ["runId", "createdAt"]), - - workDeliveries: defineTable({ - artifactId: v.optional(v.id("workArtifacts")), + .index("by_organizationId_and_createdAt", ["organizationId", "createdAt"]) + .index("by_primaryProjectId_and_createdAt", [ + "primaryProjectId", + "createdAt", + ]), + threads: defineTable({ createdAt: v.number(), - externalId: v.string(), - idempotencyKey: v.string(), - kind: v.union( - v.literal("branch"), - v.literal("commit"), - v.literal("pull-request"), - v.literal("preview"), - v.literal("deployment") - ), - metadataJson: v.string(), + createdFromEventId: v.id("events"), organizationId: v.id("organizations"), - projectId: v.id("projects"), - provider: v.string(), - sourceRevision: v.string(), - status: v.union( - v.literal("recorded"), - v.literal("ready"), - v.literal("approved"), - v.literal("delivered"), - v.literal("failed"), - v.literal("cancelled") - ), - target: v.string(), - updatedAt: v.number(), - url: v.optional(v.string()), + projectId: v.optional(v.id("projects")), + purpose: v.string(), + title: v.string(), workId: v.id("works"), }) .index("by_workId_and_createdAt", ["workId", "createdAt"]) - .index("by_workId_and_idempotencyKey", ["workId", "idempotencyKey"]), - - // ----------------------------------------------------------------- - // 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") + .index("by_organizationId_and_createdAt", ["organizationId", "createdAt"]), + events: defineTable({ + actor: v.union( + v.object({ + kind: v.literal("user"), + userId: v.string(), + }), + v.object({ + kind: v.literal("system"), + service: v.string(), + }), + v.object({ + kind: v.literal("work"), + workId: v.id("works"), + }), + v.object({ + kind: v.literal("thread"), + threadId: v.id("threads"), + }), + v.object({ + kind: v.literal("tool"), + tool: v.string(), + }), + v.object({ + integration: v.string(), + kind: v.literal("integration"), + }) ), - 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")) + artifactIds: v.optional(v.array(v.id("artifacts"))), + causationId: v.optional(v.id("events")), + correlationId: v.string(), + idempotencyKey: v.string(), + occurredAt: v.number(), + organizationId: v.id("organizations"), + payload: v.any(), + projectId: v.optional(v.id("projects")), + recordedAt: v.number(), + replyToEventId: v.optional(v.id("events")), + scopeKind: v.union(v.literal("global"), v.literal("work")), + threadId: v.optional(v.id("threads")), + type: v.string(), + visibility: v.union( + v.literal("timeline"), + v.literal("compact"), + v.literal("internal") ), - updatedAt: v.number(), + workId: v.optional(v.id("works")), }) - .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({ - attemptId: v.string(), - createdAt: v.number(), - submissionId: v.string(), - }) - .index("by_submissionId_and_attemptId", ["submissionId", "attemptId"]) - .index("by_submissionId", ["submissionId"]), - - // Conversation-stream metadata: one row per stream path. - flueConversationStreams: defineTable({ - closed: v.boolean(), - createdAt: v.number(), - identityJson: v.string(), - incarnation: v.string(), - nextOffset: v.number(), - nextProducerSequence: v.number(), - path: v.string(), - producerEpoch: v.number(), - producerId: v.optional(v.string()), - }).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({ - appendedAt: v.number(), - attemptId: v.optional(v.string()), - path: v.string(), - producerEpoch: v.number(), - producerId: v.string(), - producerSequence: v.number(), - recordsJson: v.string(), - seq: v.number(), - submissionId: v.optional(v.string()), - }) - .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({ - closed: v.boolean(), - createdAt: v.number(), - nextSeq: v.number(), - path: v.string(), - }).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({ - appendedAt: v.number(), - dataJson: v.string(), - onceKey: v.optional(v.string()), - path: v.string(), - seq: v.number(), - }) - .index("by_path_and_seq", ["path", "seq"]) - .index("by_path_and_onceKey", ["path", "onceKey"]), - - // Workflow run records. - flueRuns: defineTable({ - durationMs: v.optional(v.number()), - endedAt: v.optional(v.string()), - errorJson: v.optional(v.string()), - inputJson: v.optional(v.string()), - isError: v.optional(v.boolean()), - resultJson: v.optional(v.string()), - runId: v.string(), - startedAt: v.string(), - status: v.union( - v.literal("active"), - v.literal("completed"), - v.literal("errored") - ), - traceCarrierJson: v.optional(v.string()), - workflowName: 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_organizationId_and_scopeKind_and_recordedAt", [ + "organizationId", + "scopeKind", + "recordedAt", ]) - .index("by_status_startedAt_runId", ["status", "startedAt", "runId"]) - .index("by_status_workflowName_startedAt_runId", [ - "status", - "workflowName", - "startedAt", - "runId", + .index("by_organizationId_and_scopeKind_and_workId_and_recordedAt", [ + "organizationId", + "scopeKind", + "workId", + "recordedAt", + ]) + .index("by_organizationId_and_projectId_and_recordedAt", [ + "organizationId", + "projectId", + "recordedAt", + ]) + .index("by_organizationId_and_threadId_and_recordedAt", [ + "organizationId", + "threadId", + "recordedAt", + ]) + .index("by_organizationId_and_causationId", [ + "organizationId", + "causationId", + ]) + .index("by_organizationId_and_idempotencyKey", [ + "organizationId", + "idempotencyKey", ]), - - // Immutable attachment bytes. Identity is (streamPath, attachmentId); reads - // are additionally scoped by conversationId. - flueAttachments: defineTable({ - attachmentId: v.string(), - bytes: v.bytes(), - conversationId: v.string(), + artifacts: defineTable({ + card: v.any(), + content: v.any(), createdAt: v.number(), - digest: v.string(), - filename: v.optional(v.string()), - mimeType: v.string(), - size: v.number(), - streamPath: v.string(), + createdByEventId: v.id("events"), + kind: v.union( + v.literal("project_setup"), + v.literal("summary"), + v.literal("plan"), + v.literal("preview"), + v.literal("blocker"), + v.literal("report") + ), + logicalKey: v.string(), + organizationId: v.id("organizations"), + projectId: v.optional(v.id("projects")), + status: v.union( + v.literal("working"), + v.literal("ready"), + v.literal("blocked"), + v.literal("failed"), + v.literal("superseded") + ), + summary: v.string(), + supersedesArtifactId: v.optional(v.id("artifacts")), + threadId: v.optional(v.id("threads")), + title: v.string(), + version: v.number(), + workId: v.optional(v.id("works")), }) - .index("by_streamPath", ["streamPath"]) - .index("by_streamPath_and_attachmentId", ["streamPath", "attachmentId"]) - .index("by_streamPath_and_conversationId_and_attachmentId", [ - "streamPath", - "conversationId", - "attachmentId", + .index("by_projectId_and_createdAt", ["projectId", "createdAt"]) + .index("by_workId_and_createdAt", ["workId", "createdAt"]) + .index("by_threadId_and_createdAt", ["threadId", "createdAt"]) + .index("by_organizationId_and_logicalKey_and_version", [ + "organizationId", + "logicalKey", + "version", ]), }); diff --git a/packages/backend/convex/timeline.test.ts b/packages/backend/convex/timeline.test.ts new file mode 100644 index 0000000..f1830f7 --- /dev/null +++ b/packages/backend/convex/timeline.test.ts @@ -0,0 +1,158 @@ +import { convexTest } from "convex-test"; +import { anyApi } from "convex/server"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; + +import schema from "./schema"; + +declare global { + interface ImportMeta { + readonly glob: (pattern: string) => Record Promise>; + } +} + +const modules = import.meta.glob("./**/*.ts"); +const api = anyApi; +const identityA = { tokenIdentifier: "https://convex.test|timeline-a" }; +const identityB = { tokenIdentifier: "https://convex.test|timeline-b" }; + +const newTest = () => convexTest({ modules, schema }); + +describe("timeline", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + test("persists a prompt once and creates the deterministic first Work flow", async () => { + const t = newTest(); + await t + .withIdentity(identityA) + .mutation(api.organizations.ensurePersonalOrganization, {}); + + const first = await t + .withIdentity(identityA) + .mutation(api.timeline.sendPrompt, { + clientIdempotencyKey: "prompt-1", + scope: { kind: "global" }, + text: "Improve the onboarding flow", + }); + await vi.runAllTimersAsync(); + await t.finishInProgressScheduledFunctions(); + + const duplicate = await t + .withIdentity(identityA) + .mutation(api.timeline.sendPrompt, { + clientIdempotencyKey: "prompt-1", + scope: { kind: "global" }, + text: "Improve the onboarding flow", + }); + expect(duplicate._id).toBe(first._id); + + const global = await t + .withIdentity(identityA) + .query(api.timeline.getGlobal, {}); + expect( + global.map((event: { readonly type: string }) => event.type) + ).toEqual(["work.created", "assistant.acknowledged", "client.prompt"]); + + const works = await t + .withIdentity(identityA) + .query(api.works.listRecent, {}); + expect(works).toHaveLength(1); + expect(works[0]?.title).toBe("Improve the onboarding flow"); + + const workTimeline = await t + .withIdentity(identityA) + .query(api.timeline.getWork, { workId: works[0]!._id }); + expect( + workTimeline.map((event: { readonly type: string }) => event.type) + ).toEqual(["artifact.published", "thread.started"]); + expect(workTimeline[0]?.artifacts).toHaveLength(1); + }); + + test("does not expose a Work to another organization", async () => { + const t = newTest(); + await t + .withIdentity(identityA) + .mutation(api.organizations.ensurePersonalOrganization, {}); + const prompt = await t + .withIdentity(identityA) + .mutation(api.timeline.sendPrompt, { + clientIdempotencyKey: "prompt-private", + scope: { kind: "global" }, + text: "Private work", + }); + await vi.runAllTimersAsync(); + await t.finishInProgressScheduledFunctions(); + + const works = await t + .withIdentity(identityA) + .query(api.works.listRecent, {}); + expect(works).toHaveLength(1); + await t + .withIdentity(identityB) + .mutation(api.organizations.ensurePersonalOrganization, {}); + + await expect( + t + .withIdentity(identityB) + .query(api.timeline.getWork, { workId: works[0]!._id }) + ).rejects.toThrow(/Organization membership required/u); + expect(prompt.payload).toEqual({ text: "Private work" }); + }); + + test("stores environment metadata and setup events against a Project", async () => { + const t = newTest(); + await t + .withIdentity(identityA) + .mutation(api.organizations.ensurePersonalOrganization, {}); + const project = await t.mutation(api.projects.persistPublicGitImport, { + remote: { documents: [], warnings: [] }, + source: { + host: "github.com", + normalizedUrl: "https://github.com/example/setup-project", + projectName: "setup-project", + repositoryPath: "example/setup-project", + url: "https://github.com/example/setup-project.git", + }, + userId: identityA.tokenIdentifier, + }); + + await t + .withIdentity(identityA) + .mutation(api.projectEnvironmentVariables.upsert, { + category: "required", + name: "DATABASE_URL", + projectId: project.id, + secretReference: "project-secret-database-url", + source: "detected", + status: "missing", + }); + await t.mutation(api.projects.appendSetupEvent, { + correlationId: "setup-1", + idempotencyKey: "setup-event-1", + payload: { missing: ["DATABASE_URL"] }, + projectId: project.id, + type: "project.setup.blocked", + visibility: "timeline", + }); + + const variables = await t + .withIdentity(identityA) + .query(api.projectEnvironmentVariables.list, { projectId: project.id }); + expect(variables).toMatchObject([ + { name: "DATABASE_URL", status: "missing" }, + ]); + expect(variables[0]).not.toHaveProperty("secretValue"); + + const setup = await t + .withIdentity(identityA) + .query(api.projects.getSetup, { projectId: project.id }); + expect( + setup.events.map((event: { readonly type: string }) => event.type) + ).toEqual(["project.setup.blocked"]); + }); +}); diff --git a/packages/backend/convex/timeline.ts b/packages/backend/convex/timeline.ts new file mode 100644 index 0000000..cf32b4e --- /dev/null +++ b/packages/backend/convex/timeline.ts @@ -0,0 +1,368 @@ +import { makeFunctionReference } from "convex/server"; +import type { FunctionReference } from "convex/server"; +import { ConvexError, v } from "convex/values"; + +import type { Doc, Id } from "./_generated/dataModel"; +import type { MutationCtx, QueryCtx } from "./_generated/server"; +import { internalMutation, mutation, query } from "./_generated/server"; +import { requireCurrentOrganization, requireWorkMember } from "./authz"; + +type TimelineEventView = Doc<"events"> & { + readonly artifacts: readonly Doc<"artifacts">[]; +}; + +const promptHandlerRef = makeFunctionReference< + "mutation", + { eventId: Id<"events"> }, + undefined +>("timeline:handlePrompt") as unknown as FunctionReference< + "mutation", + "internal", + { eventId: Id<"events"> }, + undefined +>; + +const promptPayload = (event: Doc<"events">): { readonly text: string } => { + if ( + typeof event.payload !== "object" || + event.payload === null || + !("text" in event.payload) || + typeof event.payload.text !== "string" + ) { + throw new ConvexError("Prompt event payload is invalid"); + } + return { text: event.payload.text }; +}; + +const appendEvent = async ( + ctx: MutationCtx, + input: { + readonly actor: Doc<"events">["actor"]; + readonly artifactIds?: readonly Id<"artifacts">[]; + readonly causationId?: Id<"events">; + readonly correlationId: string; + readonly idempotencyKey?: string; + readonly organizationId: Id<"organizations">; + readonly payload: unknown; + readonly projectId?: Id<"projects">; + readonly replyToEventId?: Id<"events">; + readonly scopeKind: "global" | "work"; + readonly threadId?: Id<"threads">; + readonly type: string; + readonly visibility: "timeline" | "compact" | "internal"; + readonly workId?: Id<"works">; + } +): Promise> => + await ctx.db.insert("events", { + actor: input.actor, + artifactIds: input.artifactIds ? [...input.artifactIds] : undefined, + causationId: input.causationId, + correlationId: input.correlationId, + idempotencyKey: + input.idempotencyKey ?? + `${input.type}:${input.causationId ?? input.correlationId}`, + occurredAt: Date.now(), + organizationId: input.organizationId, + payload: input.payload, + projectId: input.projectId, + recordedAt: Date.now(), + replyToEventId: input.replyToEventId, + scopeKind: input.scopeKind, + threadId: input.threadId, + type: input.type, + visibility: input.visibility, + workId: input.workId, + }); + +const toEventView = async ( + ctx: QueryCtx | MutationCtx, + event: Doc<"events"> +): Promise => { + const artifacts = await Promise.all( + (event.artifactIds ?? []).map((artifactId) => ctx.db.get(artifactId)) + ); + return { + ...event, + artifacts: artifacts.filter( + (artifact): artifact is Doc<"artifacts"> => artifact !== null + ), + }; +}; + +interface EventQuery { + readonly order: (direction: "asc" | "desc") => { + readonly take: (limit: number) => Promise[]>; + }; +} + +const takeVisibleEvents = async ( + baseQuery: EventQuery, + limit: number +): Promise[]> => { + const candidates = await baseQuery.order("desc").take(limit * 3); + return candidates + .filter((event) => event.visibility !== "internal") + .slice(0, limit); +}; + +export const sendPrompt = mutation({ + args: { + clientIdempotencyKey: v.string(), + projectId: v.optional(v.id("projects")), + replyToEventId: v.optional(v.id("events")), + scope: v.union( + v.object({ kind: v.literal("global") }), + v.object({ kind: v.literal("work"), workId: v.id("works") }) + ), + text: v.string(), + }, + handler: async (ctx, args): Promise => { + const { organizationId, userId } = await requireCurrentOrganization(ctx); + const text = args.text.trim(); + const clientIdempotencyKey = args.clientIdempotencyKey.trim(); + if (!text) { + throw new ConvexError("Prompt text is required"); + } + if (!clientIdempotencyKey) { + throw new ConvexError("Client idempotency key is required"); + } + + const existing = await ctx.db + .query("events") + .withIndex("by_organizationId_and_idempotencyKey", (q) => + q + .eq("organizationId", organizationId) + .eq("idempotencyKey", clientIdempotencyKey) + ) + .unique(); + if (existing) { + return await toEventView(ctx, existing); + } + + let workId: Id<"works"> | undefined; + const { scope } = args; + let { projectId } = args; + if (scope.kind === "work") { + const work = await ctx.db.get(scope.workId); + if (!work || work.organizationId !== organizationId) { + throw new ConvexError("Work not found"); + } + const { _id: resolvedWorkId, primaryProjectId } = work; + workId = resolvedWorkId; + projectId ??= primaryProjectId; + } + if (projectId) { + const project = await ctx.db.get(projectId); + if (!project || project.organizationId !== organizationId) { + throw new ConvexError("Project not found"); + } + } + + const eventId = await ctx.db.insert("events", { + actor: { kind: "user", userId }, + causationId: undefined, + correlationId: clientIdempotencyKey, + idempotencyKey: clientIdempotencyKey, + occurredAt: Date.now(), + organizationId, + payload: { text }, + projectId, + recordedAt: Date.now(), + replyToEventId: args.replyToEventId, + scopeKind: scope.kind, + type: "client.prompt", + visibility: "timeline", + workId, + }); + + await ctx.scheduler.runAfter(0, promptHandlerRef, { eventId }); + const event = await ctx.db.get(eventId); + if (!event) { + throw new ConvexError("Prompt event could not be read after insert"); + } + return await toEventView(ctx, event); + }, +}); + +export const getGlobal = query({ + args: { + before: v.optional(v.number()), + limit: v.optional(v.number()), + }, + handler: async (ctx, args): Promise => { + const { organizationId } = await requireCurrentOrganization(ctx); + const limit = Math.min(Math.max(Math.floor(args.limit ?? 50), 1), 100); + const baseQuery = ctx.db + .query("events") + .withIndex("by_organizationId_and_scopeKind_and_recordedAt", (q) => { + const scoped = q + .eq("organizationId", organizationId) + .eq("scopeKind", "global"); + return args.before === undefined + ? scoped + : scoped.lt("recordedAt", args.before); + }); + const events = await takeVisibleEvents(baseQuery, limit); + return await Promise.all(events.map((event) => toEventView(ctx, event))); + }, +}); + +export const getWork = query({ + args: { + before: v.optional(v.number()), + limit: v.optional(v.number()), + workId: v.id("works"), + }, + handler: async (ctx, args): Promise => { + const { organizationId } = await requireWorkMember(ctx, args.workId); + const limit = Math.min(Math.max(Math.floor(args.limit ?? 50), 1), 100); + const baseQuery = ctx.db + .query("events") + .withIndex( + "by_organizationId_and_scopeKind_and_workId_and_recordedAt", + (q) => { + const scoped = q + .eq("organizationId", organizationId) + .eq("scopeKind", "work") + .eq("workId", args.workId); + return args.before === undefined + ? scoped + : scoped.lt("recordedAt", args.before); + } + ); + const events = await takeVisibleEvents(baseQuery, limit); + return await Promise.all(events.map((event) => toEventView(ctx, event))); + }, +}); + +export const handlePrompt = internalMutation({ + args: { eventId: v.id("events") }, + handler: async (ctx, args): Promise => { + const prompt = await ctx.db.get(args.eventId); + if (!prompt || prompt.type !== "client.prompt") { + return; + } + const existingResponses = await ctx.db + .query("events") + .withIndex("by_organizationId_and_causationId", (q) => + q + .eq("organizationId", prompt.organizationId) + .eq("causationId", args.eventId) + ) + .collect(); + if ( + existingResponses.some((event) => event.type === "assistant.acknowledged") + ) { + return; + } + + const { text } = promptPayload(prompt); + const now = Date.now(); + await appendEvent(ctx, { + actor: { kind: "system", service: "timeline" }, + causationId: args.eventId, + correlationId: prompt.correlationId, + organizationId: prompt.organizationId, + payload: { text: "I have it." }, + projectId: prompt.projectId, + replyToEventId: args.eventId, + scopeKind: prompt.scopeKind, + threadId: prompt.threadId, + type: "assistant.acknowledged", + visibility: "timeline", + workId: prompt.workId, + }); + + const { actor, projectId, workId: promptWorkId } = prompt; + let workId = promptWorkId; + if (!workId) { + workId = await ctx.db.insert("works", { + createdAt: now, + createdBy: actor.kind === "user" ? actor.userId : "system", + createdFromEventId: args.eventId, + organizationId: prompt.organizationId, + primaryProjectId: projectId, + summary: text, + title: text.slice(0, 80), + }); + const workCreatedEventId = await appendEvent(ctx, { + actor: { kind: "system", service: "timeline" }, + causationId: args.eventId, + correlationId: prompt.correlationId, + organizationId: prompt.organizationId, + payload: { title: text.slice(0, 80), workId }, + projectId, + scopeKind: "global", + type: "work.created", + visibility: "compact", + workId, + }); + const threadId = await ctx.db.insert("threads", { + createdAt: now, + createdFromEventId: workCreatedEventId, + organizationId: prompt.organizationId, + projectId, + purpose: "Initial exploration", + title: "Initial exploration", + workId, + }); + await appendEvent(ctx, { + actor: { kind: "thread", threadId }, + causationId: workCreatedEventId, + correlationId: prompt.correlationId, + organizationId: prompt.organizationId, + payload: { title: "Initial exploration" }, + projectId, + scopeKind: "work", + threadId, + type: "thread.started", + visibility: "compact", + workId, + }); + } + + const previousArtifacts = await ctx.db + .query("artifacts") + .withIndex("by_workId_and_createdAt", (q) => q.eq("workId", workId)) + .order("desc") + .take(1); + const version = (previousArtifacts[0]?.version ?? 0) + 1; + const artifactId = await ctx.db.insert("artifacts", { + card: { + component: "summary", + label: "Summary", + status: "ready", + summary: text, + title: text.slice(0, 80), + }, + content: { + body: `# ${text.slice(0, 80)}\n\n${text}`, + type: "markdown", + }, + createdAt: now, + createdByEventId: args.eventId, + kind: "summary", + logicalKey: `work:${workId}:summary`, + organizationId: prompt.organizationId, + projectId, + status: "ready", + summary: text, + title: text.slice(0, 80), + version, + workId, + }); + await appendEvent(ctx, { + actor: { kind: "system", service: "timeline" }, + artifactIds: [artifactId], + causationId: args.eventId, + correlationId: prompt.correlationId, + organizationId: prompt.organizationId, + payload: { artifactId, title: text.slice(0, 80) }, + projectId, + scopeKind: "work", + type: "artifact.published", + visibility: "timeline", + workId, + }); + }, +}); diff --git a/packages/backend/convex/works.ts b/packages/backend/convex/works.ts new file mode 100644 index 0000000..7e04653 --- /dev/null +++ b/packages/backend/convex/works.ts @@ -0,0 +1,42 @@ +import { v } from "convex/values"; + +import type { Doc } from "./_generated/dataModel"; +import { query } from "./_generated/server"; +import { requireCurrentOrganization, requireWorkMember } from "./authz"; + +export const listRecent = query({ + args: { limit: v.optional(v.number()) }, + handler: async (ctx, args): Promise[]> => { + const { organizationId } = await requireCurrentOrganization(ctx); + const limit = Math.min(Math.max(Math.floor(args.limit ?? 50), 1), 100); + return await ctx.db + .query("works") + .withIndex("by_organizationId_and_createdAt", (q) => + q.eq("organizationId", organizationId) + ) + .order("desc") + .take(limit); + }, +}); + +export const get = query({ + args: { workId: v.id("works") }, + handler: async (ctx, args) => { + await requireWorkMember(ctx, args.workId); + const work = await ctx.db.get(args.workId); + if (!work) { + return null; + } + const threads = await ctx.db + .query("threads") + .withIndex("by_workId_and_createdAt", (q) => q.eq("workId", args.workId)) + .order("asc") + .collect(); + const artifacts = await ctx.db + .query("artifacts") + .withIndex("by_workId_and_createdAt", (q) => q.eq("workId", args.workId)) + .order("desc") + .take(50); + return { artifacts, threads, work }; + }, +}); diff --git a/packages/backend/package.json b/packages/backend/package.json index e303457..6ab90a0 100644 --- a/packages/backend/package.json +++ b/packages/backend/package.json @@ -9,15 +9,14 @@ "dev:once": "convex dev --once --env-file=../../.env", "dev:setup": "convex dev --env-file ../../.env --configure --until-success", "check-types": "tsc --noEmit -p convex/tsconfig.json", - "test": "CONVEX_SITE_URL=https://convex.test FLUE_DB_TOKEN=test-token SITE_URL=https://site.test vitest run", - "test:watch": "CONVEX_SITE_URL=https://convex.test FLUE_DB_TOKEN=test-token SITE_URL=https://site.test vitest" + "test": "CONVEX_SITE_URL=https://convex.test SITE_URL=https://site.test vitest run", + "test:watch": "CONVEX_SITE_URL=https://convex.test SITE_URL=https://site.test vitest" }, "dependencies": { "@better-auth/expo": "catalog:", "@code/env": "workspace:*", "@code/primitives": "workspace:*", "@convex-dev/better-auth": "catalog:", - "@convex-dev/workflow": "0.4.4", "better-auth": "catalog:", "convex": "catalog:", "effect": "catalog:",