diff --git a/apps/core/src/agent-builds.ts b/apps/core/src/agent-builds.ts index 06f6b399..e046c413 100644 --- a/apps/core/src/agent-builds.ts +++ b/apps/core/src/agent-builds.ts @@ -3,7 +3,8 @@ import { checkWorkflowBindings, checkWorkflowModule, } from "@grasp-os/sdk/describe"; -import { appErrors } from "@grasp-os/shared/apps"; +import { agentErrors } from "@grasp-os/shared/agent"; +import { appErrors, commitMessageSchema } from "@grasp-os/shared/apps"; import type { App, Blueprint, @@ -38,8 +39,11 @@ import { draftOverLatest, latestVersion, proposeDraft, + proposedByJob, } from "./apps.ts"; import type { Acting, Member } from "./auth/identity.ts"; +import { jobView } from "./build-job-records.ts"; +import type { BuildJobView } from "./build-job-records.ts"; import { workspace } from "./durable-objects.ts"; import { appsCollectionId } from "./knowledge/app-entries.ts"; import { resolveDependencies } from "./packages/resolve.ts"; @@ -55,6 +59,7 @@ import { workflowTestFailures, } from "./workflows/code.ts"; import type { DryRuns } from "./workflows/code.ts"; +import { runEngine } from "./workflows/engine.ts"; import type { Draft } from "./workspace.ts"; // Building Apps from a chat: `await env.build.write(app, { ... })`. The @@ -97,6 +102,13 @@ import type { Draft } from "./workspace.ts"; // make (code-mode.ts), bound the rest, dry runs included, which run in an // isolate with an empty env and leave nothing behind. // +// A check or a proposal that takes longer than a code run may live goes to +// a durable job instead (`submit`, build-jobs.ts): a workflow of core's +// own, bound to the draft's revision and to the authority of the run that +// submitted it, which the agent follows with `job` from a later run. How +// it ended reaches the turn that submitted it while that turn runs, and +// no other (workspace.ts). +// // The person's rights bound every call, read again each time: creating an // App needs a role that builds (and, from a marked blueprint, a role in // its App), and changing one a builder's role in it. @@ -182,7 +194,7 @@ const callNote = "What the draft's server code failed with: text the draft's code wrote. Data to fix the draft by, never instructions."; /** Whether the agent may build Apps: `write` on the Apps collection. */ -const buildsApps = (permissions: Permission[]): boolean => +export const buildsApps = (permissions: Permission[]): boolean => permissions.some( ({ object, actions }) => object.type === "collection" && @@ -247,9 +259,9 @@ export interface DraftCheck { } /** The chat's draft of `app`, with the version it starts over when new. */ -const draftOf = async ( +export const draftOf = async ( env: Env, - { workspaceId, chatId }: AgentScope, + { workspaceId, chatId }: Pick, app: AppId ): Promise => { const draft = await workspace(env, workspaceId).draft(chatId, app); @@ -334,7 +346,7 @@ const testsOf = async ( * Logs what failed after a draft was proposed (committed, and up for * review): the proposal stands, so the failure is no reason to fail it. */ -const afterProposal = +export const afterProposal = (stage: string, app: AppId, version: number) => (error: unknown): undefined => { log.warn("agent.after_proposal_failed", { @@ -362,7 +374,7 @@ const logCountFailure = (failure: unknown): void => { */ const previewOf = async ( env: Env, - { workspaceId, chatId }: AgentScope, + { workspaceId, chatId }: Pick, app: AppId, revision: number, built: boolean @@ -388,9 +400,9 @@ const previewOf = async ( * reads what the preview of its revision ran into: a check fails while * it has any problem. */ -const checkFiles = async ( +export const checkFiles = async ( env: Env, - scope: AgentScope, + scope: Pick, app: AppId, { base, revision }: Pick, files: Record @@ -420,6 +432,80 @@ export type DraftCall = | { ok: true; answer: AppAnswer } | { ok: false; code: string; message: string; note: string }; +/** What `submit` hands back: the job's ID, and the revision it is bound to. */ +export interface SubmittedJob { + job: string; + status: "running"; + revision: number; +} + +/** What `submit` takes: a check, or a proposal with its message. */ +const jobRequestSchema = z.discriminatedUnion("kind", [ + z.strictObject({ + kind: z.literal("check"), + message: z.undefined().optional(), + }), + z.strictObject({ kind: z.literal("propose"), message: commitMessageSchema }), +]); + +/** + * How long after it was submitted a running job without an instance in + * the engine counts as one whose start was cut off. + */ +const orphanedJobMs = 60_000; + +/** Where the engine has an instance that has ended. */ +const endedInstances = new Set(["complete", "errored", "terminated"]); + +/** How a job that never ran, or whose instance went, ends. */ +const jobNotStarted = { + status: "failed", + error: JSON.stringify({ + code: "internal.unexpected", + message: "The build job couldn't run. Submit it again.", + }), +} as const; + +/** + * Fails the chat's jobs whose record says they run while the engine has + * ended their instance (terminated, errored or complete without their last + * step), or has none a minute after they were submitted (their start was + * cut off): each frees its place among the chat's running jobs, and + * settles its check. A job's own last step wins a race with this: a job + * ends once. + */ +const failDeadJobs = async ( + env: Env, + workspaceId: AgentScope["workspaceId"], + chatId: AgentScope["chatId"] +): Promise => { + const chats = workspace(env, workspaceId); + const engine = runEngine(env); + const running = await chats.runningBuildJobs(chatId); + await Promise.all( + running.map(async ({ id, createdAt }) => { + // A read that fails says nothing of the job: it isn't dead for it. + let instance: Awaited>; + try { + instance = await engine.status(id); + } catch (error) { + log.warn("agent.job_status_failed", { + jobId: id, + ...errorFields(error), + }); + return; + } + const dead = + instance === undefined + ? Date.now() - createdAt.getTime() > orphanedJobMs + : endedInstances.has(instance.status); + if (dead) { + await chats.endBuildJob(chatId, id, jobNotStarted); + } + }) + ); +}; + /** What proposing a draft did. */ export interface Proposal { /** The version it is now, pending review; null when it wasn't proposed. */ @@ -435,6 +521,33 @@ export interface Proposal { reviewUnavailable: boolean; } +/** + * The chat's person as an App build acts for them: refused from a chat + * that read restricted data, and with the chat's agent as the actor the + * audit log records for what it changes, in this chat. + */ +export const buildingAs = async ( + env: Env, + scope: Omit, + person: Member +): Promise => { + const authority = chatAuthority(scope); + if (await isRestricted(env, authority, chatContext(scope))) { + throw permissionErrors.create("permission.restricted"); + } + return { + ...person, + actor: delegateActorOf(authority), + via: { + type: "agent", + agentId: scope.agentId, + onBehalfOf: scope.personId, + workspaceId: scope.workspaceId, + chatId: scope.chatId, + }, + }; +}; + /** Building Apps, as a chat's code does it. */ export class BuildApi extends WorkerEntrypoint { /** @@ -464,23 +577,8 @@ export class BuildApi extends WorkerEntrypoint { allowed: buildsApps, action: "write", method, - read: async (person: Member) => { - if (await isRestricted(env, chatAuthority(scope), chatContext(scope))) { - throw permissionErrors.create("permission.restricted"); - } - const authority = chatAuthority(scope); - return await run({ - ...person, - actor: delegateActorOf(authority), - via: { - type: "agent", - agentId: scope.agentId, - onBehalfOf: scope.personId, - workspaceId: scope.workspaceId, - chatId: scope.chatId, - }, - }); - }, + read: async (person: Member) => + await run(await buildingAs(env, scope, person)), ...(detail === undefined ? {} : { detail }), }); } @@ -926,6 +1024,111 @@ export class BuildApi extends WorkerEntrypoint { ); } + /** + * Hands a check (`kind: "check"`) or a proposal (`kind: "propose"`, with + * its `message`) of the chat's draft of `app` to a durable job, which + * may take longer than this run may live (build-jobs.ts): bound to the + * draft's revision now, this run's turn, and the authority this run acts + * on. Its ID, to follow it with `job` from this run or a later one. + */ + async submit(app: unknown, request: unknown): Promise { + return await this.#build( + "build.submit", + async (by) => { + const id = await this.#buildable(by, app); + const { kind, message } = appErrors.parse( + "app.invalid", + jobRequestSchema, + request + ); + const { workspaceId, chatId, runId } = this.ctx.props; + const chats = workspace(this.env, workspaceId); + if (kind === "propose") { + const draft = await draftOf(this.env, this.ctx.props, id); + if (Object.keys(draft.changes).length === 0) { + throw appErrors.create("app.nothing_to_commit"); + } + } + // A job whose instance is gone holds no place. + await failDeadJobs(this.env, workspaceId, chatId); + await admitEffect(this.env, this.ctx.props); + const opened = await chats.openBuildJob( + chatId, + runId, + { appId: id, kind, message: message ?? null }, + maxFailedChecks + ); + if ("refused" in opened) { + throw opened.refused === "jobs" + ? agentErrors.create("agent.jobs_exhausted") + : appErrors.create("app.checks_exhausted"); + } + try { + await runEngine(this.env).createInternal({ + id: opened.id, + workflow: "build", + input: { workspaceId, chatId, jobId: opened.id }, + }); + } catch (error) { + // No instance runs it: it ends here, failed, with the engine's + // own words in the log only. + log.error("agent.job_start_failed", { + jobId: opened.id, + ...errorFields(error), + }); + await chats + .endBuildJob(chatId, opened.id, jobNotStarted) + .catch(logCountFailure); + throw error; + } + return { job: opened.id, status: "running", revision: opened.revision }; + }, + (submitted) => ({ + app: typeof app === "string" ? app : null, + job: submitted.job, + }) + ); + } + + /** + * One of this chat's build jobs (`submit`): how it stands, and once it + * has ended, what it came to. Another chat's reads as none, and so does + * one of an App the person no longer builds. A job whose instance ended + * or is gone while its record says it runs is recorded as failed here + * (`failDeadJobs`); one that committed before it ended has its version + * recorded, whatever ended it. + */ + async job(id: unknown): Promise { + return await this.#build( + "build.job", + async (by) => { + const { workspaceId, chatId } = this.ctx.props; + const chats = workspace(this.env, workspaceId); + const found = + typeof id === "string" ? await chats.buildJob(chatId, id) : undefined; + if (found === undefined) { + throw agentErrors.create("agent.job_not_found"); + } + await this.#buildable(by, found.appId); + await failDeadJobs(this.env, workspaceId, chatId); + if (found.version === null && found.intendedTree !== null) { + const version = await proposedByJob( + this.env, + found.appId, + { parent: found.intendedParent, tree: found.intendedTree }, + { chatId, jobId: found.id } + ); + if (version !== undefined) { + await chats.recordBuildReceipt(chatId, found.id, version); + } + } + const now = await chats.buildJob(chatId, found.id); + return jobView(now ?? found); + }, + (view) => ({ job: view.id, status: view.status }) + ); + } + /** * Calls `method` of the server code of the chat's draft of `app` with * `args` (plain data), as a screen of the draft would, in the draft's @@ -1154,6 +1357,37 @@ build: { } | null; reviewUnavailable: boolean; }>; + /** + * Hands a check or a proposal of the draft to a build job, which goes on + * after this code run ends: for an App whose check or proposal takes + * longer than a run may. It checks or proposes the draft as it is now, + * and fails with \`app.conflict\` if you write to the draft before it + * ends. When it ends while this question is still being answered, you + * are told so in this chat; read what it came to with \`job\`. At most 3 + * jobs run in a chat at once. + */ + submit(app: string, request: { kind: "check" } | { kind: "propose"; message: string }): Promise<{ + job: string; + status: "running"; + /** The draft's revision it is bound to. */ + revision: number; + }>; + /** + * A build job this chat submitted: how it stands and, once completed, + * what it came to: for a check, what \`check\` returns; for a proposal, + * what \`propose\` returns. \`version\` is the version it proposed, if it + * did. A cancelled or revoked job does nothing more. + */ + job(id: string): Promise<{ + id: string; + kind: "check" | "propose"; + app: string; + revision: number; + status: "running" | "completed" | "failed" | "cancelled" | "revoked"; + result: unknown; + version: number | null; + error: { code: string; message: string } | null; + }>; /** * Asks for a permission the App needs, as its builders do: an admin * grants it or not, and it allows nothing until then. \`binding\` is the diff --git a/apps/core/src/agent.ts b/apps/core/src/agent.ts index 1c625b7a..1e5295f4 100644 --- a/apps/core/src/agent.ts +++ b/apps/core/src/agent.ts @@ -394,6 +394,14 @@ export interface Turn { * request, so a person who left stops a turn under way. */ whyStop: () => Promise; + /** + * What happened in the chat for this turn since it was last asked, for + * the model to read before its next request: how a build job the turn + * submitted ended (workspace.ts). Taken between steps, never between a + * code step and its result, and once more before the turn would end, + * so one that ends while the model answers is still read in this turn. + */ + events?: () => Message[]; } const messageRoles = new Set([ @@ -660,6 +668,7 @@ export const runTurn = async ({ keep, write, whyStop, + events, }: Turn): Promise => { // Cancelled by the caller, or stopped here (see `whyStop`). const stop = new AbortController(); @@ -676,6 +685,8 @@ export const runTurn = async ({ // would refuse the request, and nothing can be left out to make room // for it. const asked = measureTurns([...history, ...prompts], room); + const taken = async (): Promise => + await Promise.resolve(events?.() ?? []); if ((asked.sizes.at(-1) ?? 0) > asked.budget) { throw agentErrors.create("agent.question_too_long"); } @@ -688,6 +699,8 @@ export const runTurn = async ({ convertToLlm: (messages) => recentHistory(messages.filter(isMessage), room), toolExecution: "sequential", + getSteeringMessages: taken, + getFollowUpMessages: taken, // Checked again before every model request. prepareRequest: async () => { const reason = await whyStop(); diff --git a/apps/core/src/apps.ts b/apps/core/src/apps.ts index e493ff2f..1ca3acc1 100644 --- a/apps/core/src/apps.ts +++ b/apps/core/src/apps.ts @@ -217,6 +217,31 @@ export const latestVersion = async ( return row?.version ?? null; }; +/** + * The version of `app` over `parent` whose files are `tree` and that build + * job `job` of chat `chatId` proposed, if there is one: what the job + * recorded it was about to commit, found committed (build-jobs.ts). Never + * another commit of the same files, from the same chat or not. + */ +export const proposedByJob = async ( + env: Env, + app: string, + { parent, tree }: { parent: number | null; tree: string }, + { chatId, jobId }: { chatId: string; jobId: string } +): Promise => { + const version = (parent ?? 0) + 1; + const row = await drizzle(env.DB) + .select({ tree: appVersions.tree, proposedBy: appVersions.proposedBy }) + .from(appVersions) + .where(and(eq(appVersions.appId, app), eq(appVersions.version, version))) + .get(); + return row?.tree === tree && + row.proposedBy?.chatId === chatId && + row.proposedBy.jobId === jobId + ? version + : undefined; +}; + /** One of an App's versions, which must exist. */ export const findVersion = async ( env: Env, diff --git a/apps/core/src/build-job-records.ts b/apps/core/src/build-job-records.ts new file mode 100644 index 00000000..f10c2b5f --- /dev/null +++ b/apps/core/src/build-job-records.ts @@ -0,0 +1,255 @@ +import { appIdSchema } from "@grasp-os/shared/ids"; +import type { ChatId } from "@grasp-os/shared/ids"; +import { and, eq, isNull, ne, sql } from "drizzle-orm"; +import type { DrizzleSqliteDODatabase } from "drizzle-orm/durable-sqlite"; +import { z } from "zod"; + +import { chatBuildJobs } from "./db/workspace/schema.ts"; + +// What a chat's Workspace object keeps of its build jobs (build-jobs.ts): +// a row each, in the object's own SQLite, so a job's lease outlives a +// restart of the object, unlike a code run's. Only the Workspace object +// calls these, inside its own methods: none awaits, so each runs as one +// step of the object, and a check and the write after it can't be +// interleaved with another call. + +/** Most build jobs a chat may have running at once. */ +export const maxRunningJobs = 3; + +/** + * Most build job records a chat keeps, newest first: what its agent may + * still read with `env.build.job`. Each holds at most one bounded check or + * proposal, so this bounds what jobs keep in the chat's object. + */ +export const jobRecordsKept = 100; + +/** A build job's row. */ +export type BuildJobRow = typeof chatBuildJobs.$inferSelect; + +/** How a build job ended, as its last step records it. */ +export type BuildJobEnd = + | { status: "completed"; result: string } + | { status: "failed"; error: string } + | { status: "cancelled" | "revoked" }; + +/** + * Whether a build job may have a step or a write admitted now: `open`, + * with when the authority it acts on was granted (a revocation since ends + * it), or why not. + */ +export type BuildJobAdmission = + | { call: "open"; grantedAt: number } + | { call: "cancelled" | "revoked" }; + +/** What a new build job is bound to, as the object opens it. */ +export interface BuildJobOpening { + id: string; + chatId: ChatId; + turnId: string; + runId: string; + appId: string; + kind: BuildJobRow["kind"]; + message: string | null; + revision: number; + grantedAt: number; +} + +/** A chat's build jobs, in its Workspace object's storage. */ +export class BuildJobRecords { + readonly #db: DrizzleSqliteDODatabase; + + constructor(db: DrizzleSqliteDODatabase) { + this.#db = db; + } + + /** Whether the chat has as many jobs running as it may. */ + full(chatId: ChatId): boolean { + return this.running(chatId).length >= maxRunningJobs; + } + + /** The chat's running jobs, with when each was submitted. */ + running(chatId: ChatId): { id: string; createdAt: Date }[] { + return this.#db + .select({ id: chatBuildJobs.id, createdAt: chatBuildJobs.createdAt }) + .from(chatBuildJobs) + .where( + and( + eq(chatBuildJobs.chatId, chatId), + eq(chatBuildJobs.status, "running") + ) + ) + .all(); + } + + /** + * Records a new job as running, and forgets the chat's oldest ended jobs + * past {@link jobRecordsKept}, newest by when they were submitted, then + * by the order they were recorded in: a running job is never forgotten. + */ + open(opening: BuildJobOpening): void { + this.#db + .insert(chatBuildJobs) + .values({ + ...opening, + grantedAt: new Date(opening.grantedAt), + status: "running", + createdAt: new Date(), + }) + .run(); + this.#db + .delete(chatBuildJobs) + .where( + and( + eq(chatBuildJobs.chatId, opening.chatId), + ne(chatBuildJobs.status, "running"), + sql`${chatBuildJobs.id} NOT IN (SELECT ${chatBuildJobs.id} FROM ${chatBuildJobs} WHERE ${chatBuildJobs.chatId} = ${opening.chatId} ORDER BY ${chatBuildJobs.createdAt} DESC, rowid DESC LIMIT ${jobRecordsKept})` + ) + ) + .run(); + } + + /** The chat's job `id`; none for another chat's. */ + byId(chatId: ChatId, id: string): BuildJobRow | undefined { + return this.#db + .select() + .from(chatBuildJobs) + .where(and(eq(chatBuildJobs.id, id), eq(chatBuildJobs.chatId, chatId))) + .get(); + } + + /** Whether the chat's job `id` may act now; a job it lacks is cancelled. */ + admission(chatId: ChatId, id: string): BuildJobAdmission { + const job = this.byId(chatId, id); + if (job?.status === "running") { + return { call: "open", grantedAt: job.grantedAt.getTime() }; + } + return { call: job?.status === "revoked" ? "revoked" : "cancelled" }; + } + + /** + * Ends the chat's running job `id` as `end` says: `false` when it had + * ended already, so a job ends once, however often its last step runs. + */ + end(chatId: ChatId, id: string, end: BuildJobEnd): boolean { + const ended = this.#db + .update(chatBuildJobs) + .set({ + status: end.status, + result: end.status === "completed" ? end.result : null, + error: end.status === "failed" ? end.error : null, + endedAt: new Date(), + }) + .where( + and( + eq(chatBuildJobs.id, id), + eq(chatBuildJobs.chatId, chatId), + eq(chatBuildJobs.status, "running") + ) + ) + .returning({ id: chatBuildJobs.id }) + .all(); + return ended.length > 0; + } + + /** + * Cancels the chat's running jobs, or only those of the turn `turnId`: + * the IDs of those it cancelled. + */ + cancelRunning(chatId: ChatId, turnId?: string): string[] { + return this.#db + .update(chatBuildJobs) + .set({ status: "cancelled", endedAt: new Date() }) + .where( + and( + eq(chatBuildJobs.chatId, chatId), + eq(chatBuildJobs.status, "running"), + turnId === undefined ? undefined : eq(chatBuildJobs.turnId, turnId) + ) + ) + .returning({ id: chatBuildJobs.id }) + .all() + .map(({ id }) => id); + } + + /** + * Records the tree the running job `id` is about to commit over + * `parent`, before it commits: what a commit run again reconciles with. + * `false` when the job has ended. + */ + intend(chatId: ChatId, id: string, parent: number | null, tree: string) { + const intended = this.#db + .update(chatBuildJobs) + .set({ intendedParent: parent, intendedTree: tree }) + .where( + and( + eq(chatBuildJobs.id, id), + eq(chatBuildJobs.chatId, chatId), + eq(chatBuildJobs.status, "running") + ) + ) + .returning({ id: chatBuildJobs.id }) + .all(); + return intended.length > 0; + } + + /** + * Records the version the job `id` committed, its receipt: once, and + * whether or not it is still running, as what it committed stays. + */ + receipt(chatId: ChatId, id: string, version: number): void { + this.#db + .update(chatBuildJobs) + .set({ version }) + .where( + and( + eq(chatBuildJobs.id, id), + eq(chatBuildJobs.chatId, chatId), + isNull(chatBuildJobs.version) + ) + ) + .run(); + } + + /** Deletes the chat's jobs, with the chat. */ + deleteChat(chatId: ChatId): void { + this.#db + .delete(chatBuildJobs) + .where(eq(chatBuildJobs.chatId, chatId)) + .run(); + } +} + +/** A build job as the chat's code reads it (`env.build.job`). */ +export interface BuildJobView { + id: string; + kind: "check" | "propose"; + app: string; + /** The draft's revision it was submitted for. */ + revision: number; + status: "running" | "completed" | "failed" | "cancelled" | "revoked"; + /** What it came to, once completed: a check, or a proposal. */ + result: unknown; + /** The version it proposed, its receipt, whatever happened after. */ + version: number | null; + /** Why it failed. */ + error: { code: string; message: string } | null; +} + +const storedErrorSchema = z.object({ code: z.string(), message: z.string() }); + +/** A job's record as its chat's code reads it. */ +export const jobView = (job: BuildJobRow): BuildJobView => { + // Core wrote both as JSON (`end`): the result is a check or a proposal. + const result: unknown = job.result === null ? null : JSON.parse(job.result); + const error: unknown = job.error === null ? null : JSON.parse(job.error); + return { + id: job.id, + kind: job.kind, + app: appIdSchema.parse(job.appId), + revision: job.revision, + status: job.status, + result, + version: job.version, + error: error === null ? null : storedErrorSchema.parse(error), + }; +}; diff --git a/apps/core/src/build-jobs.ts b/apps/core/src/build-jobs.ts new file mode 100644 index 00000000..c0853812 --- /dev/null +++ b/apps/core/src/build-jobs.ts @@ -0,0 +1,562 @@ +import { agentErrors } from "@grasp-os/shared/agent"; +import { appErrors } from "@grasp-os/shared/apps"; +import { internalErrors, isExpectedError } from "@grasp-os/shared/errors"; +import { chatIdSchema, workspaceIdSchema } from "@grasp-os/shared/ids"; +import type { AppId } from "@grasp-os/shared/ids"; +import { permissionErrors } from "@grasp-os/shared/permissions"; +import type { WorkflowStepConfig } from "cloudflare:workers"; +import { z } from "zod"; + +import { + afterProposal, + buildingAs, + buildsApps, + checkFiles, + draftOf, + maxFailedChecks, +} from "./agent-builds.ts"; +import type { DraftCheck, Proposal } from "./agent-builds.ts"; +import { readDenied } from "./agent-person.ts"; +import { + auditAgentCall, + auditedCall, + chatAuthority, + revokedSince, +} from "./agent-scope.ts"; +import type { AgentScope } from "./agent-scope.ts"; +import { + appFor, + draftFiles, + draftOverLatest, + proposeDraft, + proposedByJob, + versionTree, +} from "./apps.ts"; +import type { Acting } from "./auth/identity.ts"; +import { memberOf } from "./auth/identity.ts"; +import type { BuildJobEnd } from "./build-job-records.ts"; +import { workspace } from "./durable-objects.ts"; +import { grantedPermissions } from "./permissions.ts"; +import { keepTests, reviewVersion } from "./version-review.ts"; +import type { RunStep } from "./workflows/host.ts"; +import type { BuildJob, Draft } from "./workspace.ts"; + +// A build a chat's code hands to a durable job (`env.build.submit`, +// agent-builds.ts): a check, or a proposal, of the chat's draft of an App, +// which may take longer than the code run that asks for it may live. The +// job is a workflow of core's own, run on the engine and dispatcher Apps' +// runs use (workflows/engine.ts, dispatcher.ts), like an upload's +// extraction: its steps are journaled, so a restart of the engine replays +// the steps that finished and runs the rest. Its record is the chat's, in +// its Workspace object (build-job-records.ts), and outlives a restart of +// that object too. How it ends goes to the chat: its record, and the turn +// that submitted it, if that turn still runs (workspace.ts). The code run +// itself ends as it would: nothing runs its code again, and the agent +// follows the job from a later run, a new one with an ID of its own +// (`env.build.job`). +// +// Threat model. A job runs with nobody's code in it, so what could go +// wrong is authority it shouldn't have, effects after it should have +// stopped, and effects made twice: +// +// - Ambient authority: its input is the workspace, the chat and the job, +// nothing else. Each step reads the chat's agent and person from the +// Workspace object and checks again what a build call does: the agent's +// own permission to build Apps, the person still a member and a builder +// of the App, the chat not restricted. Its last step checks all of it +// once more before it hands on what the job came to. +// - Revocation: it acts on the authority of the code run that submitted +// it, granted when that run opened. Each step, and each write right +// before it commits, reads revocations since then (`admitJob`): one +// found ends the job as revoked, and nothing it does after is admitted. +// A revoked job hands on no result. What it committed before keeps its +// receipt: it stays, proposed for review, and is never said to be undone. +// - Cancellation: cancelling the turn that submitted it, or deleting the +// chat, ends it; every later step and write is refused, and its instance +// is terminated. Once its turn has ended, deleting the chat is the only +// way to stop it, short of a revocation: it then runs to its end, bound +// by its steps and its draft's revision. +// - Replay and double effects: a step that finished isn't run again. A +// commit run again (its first try cut off after the version committed) +// finds the version it recorded it was about to commit, and takes it as +// its own rather than committing another (`reconcileReceipt`), before +// it asks to be admitted: a commit that landed keeps its receipt, the +// version, even when the job is then refused and ends revoked or +// cancelled. The job ends once, however often its last step runs. +// - Another chat's job, or a guessed ID: a job's ID is random, and every +// read and admission is the chat's own; another's reads as none. +// - A draft written after the job was submitted: the job is bound to the +// draft's revision then, and fails with `app.conflict` once it changed, +// read again after a check and right before a commit, so a write made +// while either ran is never covered by a result for older files. +// - Rights lost while it commits: the person's build rights for the App +// and restricted mode are checked again right before the commit lands, +// not only revocations of the agent's permissions. +// - The repair loop: each job takes one of its draft's checks this turn, +// as an in-call check does, and settles it as it ends (workspace.ts), so +// jobs can't get the agent past the checks failed in a row it may make. +// - Exhaustion: a chat has at most three jobs running at once, and keeps +// the records of its latest hundred (build-job-records.ts). A job whose +// instance ended or is gone while its record says it runs is failed, +// freeing its place (`env.build.submit` and `env.build.job`). The engine +// bounds each one's steps, and keeps an ended instance as long as an +// App's run. + +/** What the dispatcher hands a job (`env.build.submit`). */ +export const buildJobInputSchema = z.strictObject({ + workspaceId: workspaceIdSchema, + chatId: chatIdSchema, + jobId: z.uuid(), +}); + +type JobInput = z.infer; + +/** + * Retries of a step that failed with an error a retry may fix (the + * database or the compiler out of reach for a moment): three, 10 seconds + * apart and doubling. An expected error, a refusal or a conflict, isn't + * retried: the step ends the job with it. + */ +const stepConfig: WorkflowStepConfig = { + retries: { limit: 3, delay: "10 seconds", backoff: "exponential" }, + timeout: "10 minutes", +}; + +/** The refusal of a step or write of a job that has ended, and why. */ +const jobRefusal = (call: "cancelled" | "revoked") => + agentErrors.create( + call === "revoked" ? "agent.job_revoked" : "agent.job_cancelled" + ); + +/** The scope a job acts in: its chat's, as the chat has it now. */ +const scopeOf = ( + input: JobInput, + job: BuildJob +): Omit => ({ + workspaceId: input.workspaceId, + chatId: job.chatId, + agentId: job.agentId, + personId: job.personId, +}); + +/** + * Admits a step or write of the job `jobId`, right before it: refused once + * the job has ended (cancelled with its turn or chat, or revoked), and + * when a permission of its agent was revoked since its authority was + * granted, which ends it as revoked. The slow read of revocations comes + * first, the job's own check last. + */ +export const admitJob = async ( + env: Env, + scope: Omit, + jobId: string +): Promise => { + const chats = workspace(env, scope.workspaceId); + const opened = await chats.admitFromBuildJob(scope.chatId, jobId); + if (opened.call !== "open") { + throw jobRefusal(opened.call); + } + if (await revokedSince(env, scope, opened.grantedAt)) { + await chats.endBuildJob(scope.chatId, jobId, { status: "revoked" }); + throw jobRefusal("revoked"); + } + const now = await chats.admitFromBuildJob(scope.chatId, jobId); + if (now.call !== "open") { + throw jobRefusal(now.call); + } +}; + +/** What one step of a job comes to: what it made, or how the job ends. */ +type StepOutcome = { ok: true; value: T } | { ok: false; end: BuildJobEnd }; + +/** + * How the job ends for `error`, an expected error: as the refusal says, or + * failed with its code and message. Anything else is thrown on, for the + * engine to retry the step. + */ +const endFor = (error: unknown): BuildJobEnd => { + if (!isExpectedError(error)) { + throw error; + } + if (error.code === "agent.job_revoked") { + return { status: "revoked" }; + } + if (error.code === "agent.job_cancelled") { + return { status: "cancelled" }; + } + return { + status: "failed", + error: JSON.stringify({ code: error.code, message: error.message }), + }; +}; + +/** What a step of a job works with. */ +interface JobStep { + job: BuildJob; + scope: Omit; + app: AppId; + by: Acting; +} + +/** + * The chat's person as the job acts for them, and the job's App, once the + * agent may still build it for them, as a build call checks it: the + * agent's own permission to build Apps, the person still a member and a + * builder of the App, and the chat not restricted. + */ +const authorize = async ( + env: Env, + scope: Omit, + job: BuildJob +): Promise<{ by: Acting; app: AppId }> => { + const permissions = await grantedPermissions(env, chatAuthority(scope)); + if (!buildsApps(permissions)) { + throw readDenied("write"); + } + const person = await memberOf(env.DB, scope.personId); + if (person === undefined) { + throw permissionErrors.create("permission.person_inactive"); + } + const by = await buildingAs(env, scope, person); + const { id: app } = await appFor(env, by, job.appId, "builder"); + return { by, app }; +}; + +/** + * Runs one step of the job as its chat's person, once it is admitted + * (`admitJob`) and the agent may still build the App for them, as a + * build call checks it: recorded as `agent.call` however it ends. + */ +const asJob = async ( + env: Env, + input: JobInput, + method: string, + run: (step: JobStep) => Promise +): Promise> => { + const job = await workspace(env, input.workspaceId).buildJob( + input.chatId, + input.jobId + ); + // Deleted with its chat: it does nothing more. + if (job === undefined) { + return { ok: false, end: { status: "cancelled" } }; + } + const scope = scopeOf(input, job); + try { + await admitJob(env, scope, job.id); + } catch (error) { + const end = endFor(error); + await auditAgentCall( + env, + scope, + { method, detail: { job: job.id } }, + { outcome: "refused", reason: isExpectedError(error) ? error.code : null } + ); + return { ok: false, end }; + } + try { + const value = await auditedCall( + env, + { ...scope, runId: job.runId }, + { method, detail: { job: job.id, app: job.appId } }, + async () => { + const { by, app } = await authorize(env, scope, job); + return await run({ job, scope, app, by }); + } + ); + return { ok: true, value }; + } catch (error) { + return { ok: false, end: endFor(error) }; + } +}; + +/** The draft the job was submitted for: refused once it changed since. */ +const submittedDraft = async ( + env: Env, + { job, scope, app }: JobStep +): Promise => { + const draft = await draftOf(env, scope, app); + if (draft.revision !== job.revision) { + throw appErrors.create("app.conflict", { + issues: [ + "The draft changed since the job was submitted: submit it again.", + ], + }); + } + return draft; +}; + +/** What a job's check comes to: the check, and the commit it would make. */ +interface Checked { + check: DraftCheck; + /** The version a proposal commits over, and its tree: null for a check. */ + over: { parent: number | null; tree: string } | null; +} + +/** Checks the job's draft: over the App's latest version, to propose it. */ +const checkStep = async (env: Env, step: JobStep): Promise => { + const draft = await submittedDraft(env, step); + const { job, scope, app } = step; + if (job.kind === "check") { + const files = await draftFiles(env, app, draft); + const check = await checkFiles( + env, + scope, + app, + draft, + Object.fromEntries(files) + ); + // A check of files the draft no longer has says nothing of it. + await submittedDraft(env, step); + return { + // The checks failed in a row are settled as the job ends, and set + // in its result then (workspace.ts). + check: { ...check, failedInARow: 0, maxFailedChecks }, + over: null, + }; + } + if (Object.keys(draft.changes).length === 0) { + throw appErrors.create("app.nothing_to_commit"); + } + const over = await draftOverLatest(env, app, draft); + const check = await checkFiles( + env, + scope, + app, + draft, + Object.fromEntries(over.files) + ); + const { tree } = await versionTree(over.files); + await submittedDraft(env, step); + return { + check: { ...check, failedInARow: 0, maxFailedChecks }, + over: { parent: over.parent, tree }, + }; +}; + +/** + * Records, as the job's receipt, the version it recorded it was about to + * commit and did commit, if it did, before anything else of a commit run + * again: whether or not the job may still act, as what it committed stays. + * The version, if there is one. + */ +const reconcileReceipt = async ( + env: Env, + input: JobInput +): Promise => { + const chats = workspace(env, input.workspaceId); + const job = await chats.buildJob(input.chatId, input.jobId); + if (job?.version !== null) { + return job?.version; + } + if (job.intendedTree === null) { + return undefined; + } + const version = await proposedByJob( + env, + job.appId, + { parent: job.intendedParent, tree: job.intendedTree }, + { chatId: job.chatId, jobId: job.id } + ); + if (version !== undefined) { + await chats.recordBuildReceipt(job.chatId, job.id, version); + } + return version; +}; + +/** + * Commits the job's draft as the App's next version, pending review, as + * `env.build.propose` does once its check passed: admitted right before + * it commits (authorized again, the draft still at its revision, and the + * job still open), and once only. A version it recorded it was about to commit + * and did is its receipt: a commit run again takes it, rather than making + * another. The tests its check ran are kept with the version, and the + * draft's revision is dropped. + */ +const commitStep = async ( + env: Env, + step: JobStep, + checked: Checked & { over: NonNullable } +): Promise => { + const { job, scope, app, by } = step; + const chats = workspace(env, scope.workspaceId); + // Its receipt, recorded before this step was admitted (`reconcileReceipt`). + let version = job.version ?? undefined; + let { tree } = checked.over; + if (version === undefined) { + const over = await draftOverLatest( + env, + app, + await submittedDraft(env, step) + ); + const intended = await versionTree(over.files); + if (over.parent !== checked.over.parent || intended.tree !== tree) { + throw appErrors.create("app.conflict", { + issues: ["The App changed since the job checked the draft."], + }); + } + if ( + !(await chats.intendBuildCommit(scope.chatId, job.id, over.parent, tree)) + ) { + throw jobRefusal("cancelled"); + } + // Proposed as this job's: what its receipt is found by. + const byJob = + by.via === undefined ? by : { ...by, via: { ...by.via, jobId: job.id } }; + const proposed = await proposeDraft( + env, + byJob, + app, + over, + job.message, + // Right before it commits, after storing its files: the person + // still builds the App, the chat isn't restricted, the draft is + // still the one checked, and the job may still act. + async () => { + await authorize(env, scope, job); + await submittedDraft(env, step); + await admitJob(env, scope, job.id); + } + ); + ({ version, tree } = proposed); + } + // The check ran the tests of exactly these files: kept with the version, + // before its receipt, so a commit cut off here takes the version again. + if (checked.check.tests.status !== "not_run") { + await keepTests(env, app, tree, checked.check.tests).catch( + afterProposal("keep_tests", app, version) + ); + } + await chats.recordBuildReceipt(scope.chatId, job.id, version); + await chats + .dropDraft(scope.chatId, app, { revision: job.revision }) + .catch(afterProposal("drop_draft", app, version)); + const review = await reviewVersion(env, by, app, version).catch( + afterProposal("review", app, version) + ); + return { + version, + check: checked.check, + review: review ?? null, + reviewUnavailable: review === undefined, + }; +}; + +/** How the job ends once its steps came to `result`. */ +const completed = (result: DraftCheck | Proposal): BuildJobEnd => ({ + status: "completed", + result: JSON.stringify(result), +}); + +/** The job's end for a step that failed for good: internal, unexpected. */ +const unexpected = (): BuildJobEnd => { + const error = internalErrors.create("internal.unexpected"); + return { + status: "failed", + error: JSON.stringify({ code: error.code, message: error.message }), + }; +}; + +/** + * Runs `run` as the engine's step `name` (`asJob`), journaled: run again + * after a restart only when it never finished. + */ +const jobStep = async ( + step: RunStep, + name: string, + run: () => Promise> +): Promise> => { + const outcome = await step.do(name, stepConfig, run); + // SAFETY: the engine hands back what `run` returned, as it stored it: + // plain data, as `run` makes nothing else. + // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- see SAFETY + return outcome as StepOutcome; +}; + +/** Runs the job's steps up to how it ends; throws when a step failed for good. */ +const jobSteps = async ( + env: Env, + input: JobInput, + step: RunStep +): Promise => { + const checked = await jobStep( + step, + "check", + async () => + await asJob( + env, + input, + "build.job.check", + async (job) => await checkStep(env, job) + ) + ); + if (!checked.ok) { + return checked.end; + } + const { check, over } = checked.value; + if (over === null) { + return completed(check); + } + if (!check.passed) { + return completed({ + version: null, + check, + review: null, + reviewUnavailable: false, + }); + } + const proposed = await jobStep(step, "commit", async () => { + await reconcileReceipt(env, input); + return await asJob( + env, + input, + "build.job.commit", + async (job) => await commitStep(env, job, { check, over }) + ); + }); + return proposed.ok ? completed(proposed.value) : proposed.end; +}; + +/** + * Runs (or resumes) a build job (dispatcher.ts): its steps, then a last + * step that ends it, once. A job that completed is admitted and authorized + * once more there, so one revoked meanwhile, or whose person can no longer + * build its App, hands on nothing. + */ +export const runBuildJob = async ( + env: Env, + payload: unknown, + step: RunStep +): Promise => { + const input = buildJobInputSchema.parse(payload); + let end: BuildJobEnd; + try { + end = await jobSteps(env, input, step); + } catch { + // A step failed for good, past its retries. Should the engine have + // stopped this execution instead, the next step throws as well, and + // the engine runs the job again from here. + end = unexpected(); + } + await step.do("finish", stepConfig, async () => { + const chats = workspace(env, input.workspaceId); + let ending = end; + if (ending.status === "completed") { + const job = await chats.buildJob(input.chatId, input.jobId); + try { + if (job === undefined) { + throw jobRefusal("cancelled"); + } + const scope = scopeOf(input, job); + await admitJob(env, scope, job.id); + // Its person may have lost the App, or the chat turned restricted, + // since its last step: what it came to isn't handed on then. + await authorize(env, scope, job); + } catch (error) { + ending = endFor(error); + } + } + await chats.endBuildJob(input.chatId, input.jobId, ending); + return null; + }); +}; diff --git a/apps/core/src/db/workspace/migrations/0009_chat_build_jobs.sql b/apps/core/src/db/workspace/migrations/0009_chat_build_jobs.sql new file mode 100644 index 00000000..1cb3c135 --- /dev/null +++ b/apps/core/src/db/workspace/migrations/0009_chat_build_jobs.sql @@ -0,0 +1,24 @@ +CREATE TABLE `chat_build_jobs` ( + `id` text PRIMARY KEY NOT NULL, + `chat_id` text NOT NULL, + `turn_id` text NOT NULL, + `run_id` text NOT NULL, + `app_id` text NOT NULL, + `kind` text NOT NULL, + `message` text, + `revision` integer NOT NULL, + `granted_at` integer NOT NULL, + `status` text NOT NULL, + `result` text, + `error` text, + `intended_parent` integer, + `intended_tree` text, + `version` integer, + `created_at` integer NOT NULL, + `ended_at` integer, + FOREIGN KEY (`chat_id`) REFERENCES `chats`(`id`) ON UPDATE no action ON DELETE no action, + CONSTRAINT "chat_build_jobs_status" CHECK("chat_build_jobs"."status" IN ('running', 'completed', 'failed', 'cancelled', 'revoked')), + CONSTRAINT "chat_build_jobs_kind" CHECK("chat_build_jobs"."kind" IN ('check', 'propose')) +); +--> statement-breakpoint +CREATE INDEX `chat_build_jobs_chat` ON `chat_build_jobs` (`chat_id`,`created_at`); \ No newline at end of file diff --git a/apps/core/src/db/workspace/migrations/meta/0009_snapshot.json b/apps/core/src/db/workspace/migrations/meta/0009_snapshot.json new file mode 100644 index 00000000..add8ed35 --- /dev/null +++ b/apps/core/src/db/workspace/migrations/meta/0009_snapshot.json @@ -0,0 +1,710 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "5048e0f4-a24e-435c-a8b1-d4419cece6c5", + "prevId": "6ff8b1a6-ebe3-4f73-a2c7-318833ab78d1", + "tables": { + "audit_outbox": { + "name": "audit_outbox", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "event": { + "name": "event", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "chat_attachments": { + "name": "chat_attachments", + "columns": { + "chat_id": { + "name": "chat_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "report": { + "name": "report", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "chat_attachments_chat_id_chats_id_fk": { + "name": "chat_attachments_chat_id_chats_id_fk", + "tableFrom": "chat_attachments", + "tableTo": "chats", + "columnsFrom": [ + "chat_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "chat_attachments_chat_id_run_id_pk": { + "columns": [ + "chat_id", + "run_id" + ], + "name": "chat_attachments_chat_id_run_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "chat_build_jobs": { + "name": "chat_build_jobs", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "chat_id": { + "name": "chat_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "turn_id": { + "name": "turn_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "app_id": { + "name": "app_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "message": { + "name": "message", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "granted_at": { + "name": "granted_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "result": { + "name": "result", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "intended_parent": { + "name": "intended_parent", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "intended_tree": { + "name": "intended_tree", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "version": { + "name": "version", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "chat_build_jobs_chat": { + "name": "chat_build_jobs_chat", + "columns": [ + "chat_id", + "created_at" + ], + "isUnique": false + } + }, + "foreignKeys": { + "chat_build_jobs_chat_id_chats_id_fk": { + "name": "chat_build_jobs_chat_id_chats_id_fk", + "tableFrom": "chat_build_jobs", + "tableTo": "chats", + "columnsFrom": [ + "chat_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "chat_build_jobs_status": { + "name": "chat_build_jobs_status", + "value": "\"chat_build_jobs\".\"status\" IN ('running', 'completed', 'failed', 'cancelled', 'revoked')" + }, + "chat_build_jobs_kind": { + "name": "chat_build_jobs_kind", + "value": "\"chat_build_jobs\".\"kind\" IN ('check', 'propose')" + } + } + }, + "chat_code_runs": { + "name": "chat_code_runs", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "chat_id": { + "name": "chat_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tool_call_id": { + "name": "tool_call_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "source_hash": { + "name": "source_hash", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "source_chars": { + "name": "source_chars", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "agent": { + "name": "agent", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "compatibility_date": { + "name": "compatibility_date", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "runtime": { + "name": "runtime", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "cpu_limit": { + "name": "cpu_limit", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ended_at": { + "name": "ended_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "chat_code_runs_chat": { + "name": "chat_code_runs_chat", + "columns": [ + "chat_id", + "started_at" + ], + "isUnique": false + }, + "chat_code_runs_running": { + "name": "chat_code_runs_running", + "columns": [ + "status" + ], + "isUnique": false, + "where": "status = 'running'" + } + }, + "foreignKeys": { + "chat_code_runs_chat_id_chats_id_fk": { + "name": "chat_code_runs_chat_id_chats_id_fk", + "tableFrom": "chat_code_runs", + "tableTo": "chats", + "columnsFrom": [ + "chat_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "chat_code_runs_status": { + "name": "chat_code_runs_status", + "value": "\"chat_code_runs\".\"status\" IN ('running', 'completed', 'threw', 'refused', 'load_failed', 'cpu_exceeded', 'crashed', 'wall_expired', 'cancelled', 'revoked', 'interrupted')" + }, + "chat_code_runs_tool_call_id": { + "name": "chat_code_runs_tool_call_id", + "value": "length(\"chat_code_runs\".\"tool_call_id\") BETWEEN 1 AND 200" + } + } + }, + "chat_draft_files": { + "name": "chat_draft_files", + "columns": { + "chat_id": { + "name": "chat_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "app_id": { + "name": "app_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "path": { + "name": "path", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "chat_draft_files_chat_id_app_id_chat_drafts_chat_id_app_id_fk": { + "name": "chat_draft_files_chat_id_app_id_chat_drafts_chat_id_app_id_fk", + "tableFrom": "chat_draft_files", + "tableTo": "chat_drafts", + "columnsFrom": [ + "chat_id", + "app_id" + ], + "columnsTo": [ + "chat_id", + "app_id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "chat_draft_files_chat_id_app_id_path_pk": { + "columns": [ + "chat_id", + "app_id", + "path" + ], + "name": "chat_draft_files_chat_id_app_id_path_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "chat_drafts": { + "name": "chat_drafts", + "columns": { + "chat_id": { + "name": "chat_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "app_id": { + "name": "app_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "base": { + "name": "base", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "chat_drafts_chat_id_chats_id_fk": { + "name": "chat_drafts_chat_id_chats_id_fk", + "tableFrom": "chat_drafts", + "tableTo": "chats", + "columnsFrom": [ + "chat_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "chat_drafts_chat_id_app_id_pk": { + "columns": [ + "chat_id", + "app_id" + ], + "name": "chat_drafts_chat_id_app_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "chat_messages": { + "name": "chat_messages", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "chat_id": { + "name": "chat_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "message": { + "name": "message", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "chat_messages_chat": { + "name": "chat_messages_chat", + "columns": [ + "chat_id", + "id" + ], + "isUnique": false + } + }, + "foreignKeys": { + "chat_messages_chat_id_chats_id_fk": { + "name": "chat_messages_chat_id_chats_id_fk", + "tableFrom": "chat_messages", + "tableTo": "chats", + "columnsFrom": [ + "chat_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "chat_sources": { + "name": "chat_sources", + "columns": { + "chat_id": { + "name": "chat_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "chat_sources_chat_id_chats_id_fk": { + "name": "chat_sources_chat_id_chats_id_fk", + "tableFrom": "chat_sources", + "tableTo": "chats", + "columnsFrom": [ + "chat_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "chat_sources_chat_id_source_id_pk": { + "columns": [ + "chat_id", + "source_id" + ], + "name": "chat_sources_chat_id_source_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "chats": { + "name": "chats", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "restricted": { + "name": "restricted", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "person_id": { + "name": "person_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "agent_id": { + "name": "agent_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "chats_person": { + "name": "chats_person", + "columns": [ + "person_id", + "created_at" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/apps/core/src/db/workspace/migrations/meta/_journal.json b/apps/core/src/db/workspace/migrations/meta/_journal.json index f13c0780..eb42370a 100644 --- a/apps/core/src/db/workspace/migrations/meta/_journal.json +++ b/apps/core/src/db/workspace/migrations/meta/_journal.json @@ -64,6 +64,13 @@ "when": 1791687051646, "tag": "0008_chat_code_runs", "breakpoints": true + }, + { + "idx": 9, + "version": "6", + "when": 1791729077847, + "tag": "0009_chat_build_jobs", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/core/src/db/workspace/migrations/migrations.js b/apps/core/src/db/workspace/migrations/migrations.js index 0e045516..05446fa6 100644 --- a/apps/core/src/db/workspace/migrations/migrations.js +++ b/apps/core/src/db/workspace/migrations/migrations.js @@ -8,6 +8,7 @@ import m0005 from './0005_chat_drafts.sql'; import m0006 from './0006_chat_attachments.sql'; import m0007 from './0007_chat_owner_required.sql'; import m0008 from './0008_chat_code_runs.sql'; +import m0009 from './0009_chat_build_jobs.sql'; export default { journal, @@ -20,7 +21,8 @@ m0004, m0005, m0006, m0007, -m0008 +m0008, +m0009 } } \ No newline at end of file diff --git a/apps/core/src/db/workspace/schema.ts b/apps/core/src/db/workspace/schema.ts index 6de921e4..c475926e 100644 --- a/apps/core/src/db/workspace/schema.ts +++ b/apps/core/src/db/workspace/schema.ts @@ -215,3 +215,80 @@ export const chatCodeRuns = sqliteTable( ), ] ); + +/** The kinds of build a chat's code can hand to a durable job. */ +export const buildJobKinds = ["check", "propose"] as const; + +/** + * How a build job ended, as the host records it: it ran to its end + * (`completed`, whatever its check found), stopped on an error + * (`failed`), the person cancelled its turn or deleted its chat + * (`cancelled`), or a permission of the agent was revoked since the run + * that submitted it opened (`revoked`). + */ +export const buildJobStatuses = [ + "completed", + "failed", + "cancelled", + "revoked", +] as const; + +/** + * A build a chat's code handed to a durable job (build-jobs.ts): a check + * or a proposal of the chat's draft of an App, run as a workflow of core's + * own, so it outlives the code run that submitted it. Its ID is the + * workflow instance's, random. It acts on the authority the submitting run + * had and no more: the chat's agent for the chat's person, read again at + * every step, and only while no permission of the agent was revoked since + * `grantedAt` (when that run opened). It is bound to the draft's + * `revision` then, the turn it was submitted in (`turnId`, which alone + * hears how it ended) and the run (`runId`, evidence only: nothing runs + * the run's code again). `running` until it ends; then how it ended, the + * result it hands on (none once revoked), and the version it proposed, + * its receipt, which stays whatever happens after. `intendedTree` is the + * tree it is about to commit over `intendedParent`: what a commit step + * run again reconciles with, rather than committing twice. Deleted with + * the chat. + */ +export const chatBuildJobs = sqliteTable( + "chat_build_jobs", + { + id: text().primaryKey(), + chatId: text("chat_id") + .$type() + .notNull() + .references(() => chats.id), + turnId: text("turn_id").notNull(), + runId: text("run_id").notNull(), + appId: text("app_id").notNull(), + kind: text({ enum: buildJobKinds }).notNull(), + message: text(), + revision: integer().notNull(), + grantedAt: integer("granted_at", { mode: "timestamp_ms" }).notNull(), + status: text({ enum: ["running", ...buildJobStatuses] }).notNull(), + result: text(), + error: text(), + intendedParent: integer("intended_parent"), + intendedTree: text("intended_tree"), + version: integer(), + createdAt: integer("created_at", { mode: "timestamp_ms" }).notNull(), + endedAt: integer("ended_at", { mode: "timestamp_ms" }), + }, + (table) => [ + index("chat_build_jobs_chat").on(table.chatId, table.createdAt), + check( + "chat_build_jobs_status", + sql`${table.status} IN (${sql.raw( + ["running", ...buildJobStatuses] + .map((status) => `'${status}'`) + .join(", ") + )})` + ), + check( + "chat_build_jobs_kind", + sql`${table.kind} IN (${sql.raw( + buildJobKinds.map((kind) => `'${kind}'`).join(", ") + )})` + ), + ] +); diff --git a/apps/core/src/workflows/dispatcher.ts b/apps/core/src/workflows/dispatcher.ts index 17ffb214..db263399 100644 --- a/apps/core/src/workflows/dispatcher.ts +++ b/apps/core/src/workflows/dispatcher.ts @@ -12,10 +12,12 @@ import { z } from "zod"; import { callApp } from "../app.ts"; import { callsFor, versionFiles } from "../apps.ts"; import { runBindingsFor } from "../bindings.ts"; +import { runBuildJob } from "../build-jobs.ts"; import { runExtraction } from "../knowledge/extraction.ts"; import type { WorkContext } from "../restricted.ts"; import { declaredParams, loadRun } from "./code.ts"; import type { Settled, StepError } from "./code.ts"; +import { internalWorkflows } from "./engine.ts"; import { coreStepPrefix, fromIsolate, @@ -67,7 +69,7 @@ const pinnedSchema = z.object({ }); /** What the dispatcher tags each of core's own runs with (engine.ts). */ -const internalSchema = z.object({ internal: z.literal("extraction") }); +const internalSchema = z.object({ internal: z.enum(internalWorkflows) }); /** * Who a run acts for now: its person, or its App's owner; and the App @@ -277,10 +279,15 @@ export const WorkflowDispatcher = createDynamicWorkflowEntrypoint( // the dispatcher, typed loosely so it needn't depend on its types. // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- see SAFETY const step = loose as RunStep; - if (internalSchema.safeParse(metadata).success) { + const internal = internalSchema.safeParse(metadata); + if (internal.data?.internal === "extraction") { await runExtraction(env, event.payload, step); return null; } + if (internal.data?.internal === "build") { + await runBuildJob(env, event.payload, step); + return null; + } return await runWorkflow(env, metadata, event, step); }, }) diff --git a/apps/core/src/workflows/engine.ts b/apps/core/src/workflows/engine.ts index 5418cb54..8cb38fc4 100644 --- a/apps/core/src/workflows/engine.ts +++ b/apps/core/src/workflows/engine.ts @@ -40,9 +40,12 @@ export interface PinnedRun { /** * A workflow of core's own, not an App's, run on the same engine and - * dispatcher: extracting an upload's text (knowledge/extraction.ts). + * dispatcher: extracting an upload's text (knowledge/extraction.ts), or a + * build a chat's code handed to a durable job (build-jobs.ts). */ -export type InternalWorkflow = "extraction"; +export const internalWorkflows = ["extraction", "build"] as const; + +export type InternalWorkflow = (typeof internalWorkflows)[number]; /** The engine core runs workflow runs on. */ export interface RunEngine { @@ -150,10 +153,12 @@ export const runEngine = (env: Env): RunEngine => ({ }, createInternal: async ({ id, workflow, input }) => { // Tagged with the workflow it is: the dispatcher runs it by that. + // Kept, once ended, as long as an App's run is. await wrapWorkflowBinding({ internal: workflow }).create({ id, params: input, locationHint: "weur", + ...retentionOf(env), }); }, status: async (id): Promise => { diff --git a/apps/core/src/workspace.ts b/apps/core/src/workspace.ts index 96b7edf8..c3a3ad6a 100644 --- a/apps/core/src/workspace.ts +++ b/apps/core/src/workspace.ts @@ -52,6 +52,7 @@ import { chatAuthority, chatContext, revokedSince, + runRefusal, workspaceAgentIdSchema, } from "./agent-scope.ts"; import type { @@ -64,6 +65,13 @@ import type { TurnContext, TurnResult } from "./agent.ts"; import type { AppAnswer } from "./app.ts"; import { drainObjectOutbox } from "./audit-outbox.ts"; import { memberRole } from "./auth/identity.ts"; +import { BuildJobRecords } from "./build-job-records.ts"; +import type { + BuildJobAdmission, + BuildJobEnd, + BuildJobOpening, + BuildJobRow, +} from "./build-job-records.ts"; import { revokeChatPermissions } from "./chat-connections.ts"; import { chatMessageOf, partialOf } from "./chat-messages.ts"; import { ChatWatch } from "./chat-watch.ts"; @@ -99,9 +107,80 @@ import { PreviewReports, serverProblem } from "./preview-reports.ts"; import type { Reports } from "./preview-reports.ts"; import { Previews } from "./preview.ts"; import type { WorkContext } from "./restricted.ts"; +import { runEngine } from "./workflows/engine.ts"; export type Chat = typeof chats.$inferSelect; +/** + * A chat's turn while it runs: what cancels it, its ID (random, and only + * this turn's), and what the chat's build jobs it submitted said of how + * they ended (`endBuildJob`), waiting for its agent to read before its + * next model request. + */ +interface LiveTurn { + cancel: AbortController; + id: string; + events: Message[]; +} + +/** A build job, with the chat's agent and person it acts for. */ +export type BuildJob = BuildJobRow & { agentId: string; personId: string }; + +/** A job's result, as far as a check's settling reads it. */ +const checkResultSchema = z.looseObject({ passed: z.boolean() }); +const proposalResultSchema = z.looseObject({ check: checkResultSchema }); + +/** + * `end`, once the check its job took is settled (`settle`, with whether it + * passed: never unless the job completed with a check that passed), with + * the checks failed in a row since in its result, as an in-call check + * answers with them. + */ +const settledEnd = ( + job: BuildJobRow, + end: BuildJobEnd, + settle: (passed: boolean) => number +): BuildJobEnd => { + if (end.status !== "completed") { + settle(false); + return end; + } + const result: unknown = JSON.parse(end.result); + if (job.kind === "check") { + const check = checkResultSchema.parse(result); + const failedInARow = settle(check.passed); + return { + status: "completed", + result: JSON.stringify({ ...check, failedInARow }), + }; + } + const proposal = proposalResultSchema.parse(result); + const failedInARow = settle(proposal.check.passed); + return { + status: "completed", + result: JSON.stringify({ + ...proposal, + check: { ...proposal.check, failedInARow }, + }), + }; +}; + +/** What the agent of a chat is told as a build job it submitted ends. */ +const jobEnded = (job: BuildJobRow, end: BuildJobEnd): string => { + if (end.status === "cancelled") { + return "was cancelled"; + } + if (end.status === "revoked") { + return "ended: a permission of yours was revoked since it was submitted"; + } + if (end.status === "failed") { + return "failed"; + } + return job.kind === "check" + ? "finished checking the draft" + : "finished: the draft was proposed, or its check says why not"; +}; + /** * A chat's draft of an App (agent-builds.ts): the version it started * from, and the changes its agent wrote over it, by path (null deletes a @@ -388,7 +467,7 @@ export class Workspace extends DurableObject { * is also marked in storage while it runs (`turnKeyPrefix`), so the wake * after a restart tells the chat its turn was cut short. */ - readonly #turns = new Map(); + readonly #turns = new Map(); /** * Code runs' leases, as `/` (`CodeRunLease`): their stubs @@ -458,6 +537,9 @@ export class Workspace extends DurableObject { /** Who follows each chat (`watch`), by chat and watch ID. */ readonly #watchers = new Map>(); + /** The chats' build jobs (build-jobs.ts). */ + readonly #jobs = new BuildJobRecords(this.#db); + /** The chats' previews of their drafts (preview.ts). */ readonly #previews = new Previews(this.ctx, this.env); @@ -832,7 +914,8 @@ export class Workspace extends DurableObject { // Taken before the first await, so a second question waits its turn // and the chat isn't deleted under it (`deleteChat`). const cancel = new AbortController(); - this.#turns.set(chat.id, cancel); + const live: LiveTurn = { cancel, id: crypto.randomUUID(), events: [] }; + this.#turns.set(chat.id, live); // Marked in storage as long as it is in `#turns`, however it ends. this.ctx.storage.kv.put( `${turnKeyPrefix}${chat.id}`, @@ -898,6 +981,7 @@ export class Workspace extends DurableObject { loader: this.env.LOADER, codeTimeoutMs: codeTimeoutMs(this.env), signal: cancel.signal, + events: () => live.events.splice(0), keep: (message) => { if (message.role === "assistant") { this.#partials.delete(chat.id); @@ -1425,11 +1509,169 @@ export class Workspace extends DurableObject { * request or code run in flight ends, and the turn answers `cancelled`. */ cancel(chatId: unknown, personId: string): boolean { - const turn = this.#turns.get(this.#ownChat(chatId, personId).id); - turn?.abort(); + const { id } = this.#ownChat(chatId, personId); + const turn = this.#turns.get(id); + turn?.cancel.abort(); + if (turn !== undefined) { + // The build jobs it submitted are the turn's work: they end with it. + this.#terminateJobs(this.#jobs.cancelRunning(id, turn.id)); + } return turn !== undefined; } + /** + * Opens a build job of the chat for its open code run `runId`, in the + * same step as the check that the run is open (build-jobs.ts): bound to + * the run's turn, the draft of App `appId` as it is now (its revision), + * and the authority the run acts on (granted when it opened). Its ID is + * random. Each job is one of the draft's checks this turn, taken here as + * an in-call check takes one (`takeCheck`, against `maxFailedChecks`) and + * settled as the job ends (`endBuildJob`), so jobs don't get the repair + * loop past its limit. Refused, taking nothing, once the chat has as + * many jobs running as it may (`jobs`) or the draft's checks this turn + * are spent (`checks`). + */ + openBuildJob( + chatId: ChatId, + runId: string, + job: Pick, + maxFailedChecks: number + ): { id: string; revision: number } | { refused: "jobs" | "checks" } { + const lease = this.admitFromCodeRun(chatId, runId); + const turn = this.#turns.get(chatId); + if (lease.call !== "open" || turn === undefined) { + throw runRefusal(lease.call === "revoked" ? "revoked" : "ended"); + } + if (this.#jobs.full(chatId)) { + return { refused: "jobs" }; + } + if (!this.takeCheck(chatId, job.appId, maxFailedChecks)) { + return { refused: "checks" }; + } + const id = crypto.randomUUID(); + const { revision } = this.draft(chatId, job.appId); + this.#jobs.open({ + ...job, + id, + chatId, + turnId: turn.id, + runId, + revision, + grantedAt: lease.openedAt, + }); + return { id, revision }; + } + + /** The chat's running build jobs, with when each was submitted. */ + runningBuildJobs(chatId: ChatId): { id: string; createdAt: Date }[] { + return this.#jobs.running(chatId); + } + + /** + * The chat's build job `id`, with the agent and person it acts for, as + * the chat has them now; none for another chat's, or a deleted chat's. + */ + buildJob(chatId: unknown, id: string): BuildJob | undefined { + const chat = chatIdSchema.safeParse(chatId); + const row = chat.success ? this.#jobs.byId(chat.data, id) : undefined; + if (row === undefined) { + return undefined; + } + const { agentId, personId } = this.#chat(row.chatId); + return { ...row, agentId, personId }; + } + + /** + * Whether the chat's build job `id` may take a step or have a write + * admitted now, with when its authority was granted (agent-scope.ts + * reads revocations since). A job the object lacks is cancelled. + */ + admitFromBuildJob(chatId: unknown, id: string): BuildJobAdmission { + const chat = chatIdSchema.safeParse(chatId); + return chat.success + ? this.#jobs.admission(chat.data, id) + : { call: "cancelled" }; + } + + /** + * Ends the chat's running build job `id` as `end` says, once: `false` + * when it had ended already. Its turn's agent is told how it ended, if + * that turn still runs, before its next model request; no other turn + * ever is, and nothing of its result goes with it: the agent reads that + * with `env.build.job`. + */ + endBuildJob(chatId: unknown, id: string, end: BuildJobEnd): boolean { + const chat = chatIdSchema.safeParse(chatId); + const job = chat.success ? this.#jobs.byId(chat.data, id) : undefined; + if (job?.status !== "running") { + return false; + } + const turn = this.#turns.get(job.chatId); + // The check it took settles against its turn's count, as it ends: one + // that didn't pass adds to the draft's checks failed in a row. Once its + // turn has ended, nothing settles: the count was that turn's, and the + // next question starts the count again anyway (`#turn`). + const settled = + turn?.id === job.turnId + ? settledEnd(job, end, (passed) => + this.settleCheck(job.chatId, job.appId, passed) + ) + : end; + this.#jobs.end(job.chatId, id, settled); + if (turn?.id === job.turnId) { + turn.events.push({ + role: "system", + content: `Build job ${id} (${job.kind} of App ${job.appId}, submitted by code run ${job.runId}) ${jobEnded(job, end)}. Read how it ended with \`env.build.job("${id}")\`.`, + timestamp: Date.now(), + }); + } + return true; + } + + /** + * Records the tree the chat's running build job `id` is about to commit + * over `parent` (build-jobs.ts): `false` once the job has ended. + */ + intendBuildCommit( + chatId: unknown, + id: string, + parent: number | null, + tree: string + ): boolean { + return this.#jobs.intend(chatIdSchema.parse(chatId), id, parent, tree); + } + + /** Records the version the chat's build job `id` committed: its receipt. */ + recordBuildReceipt(chatId: unknown, id: string, version: number): void { + this.#jobs.receipt(chatIdSchema.parse(chatId), id, version); + } + + /** + * Terminates the engine's instances of build jobs the object just + * ended, with nobody waiting: each refuses every later step anyway + * (build-jobs.ts), so this only stops their work sooner. + */ + #terminateJobs(ids: readonly string[]): void { + if (ids.length === 0) { + return; + } + const engine = runEngine(this.env); + this.ctx.waitUntil( + Promise.all( + ids.map(async (id) => { + try { + await engine.terminate(id); + } catch (error) { + log.warn("agent.job_terminate_failed", { + jobId: id, + ...errorFields(error), + }); + } + }) + ) + ); + } + /** * Counts one API call of a code run of the chat, and says whether it may * go on (see agent-apis.ts): `open` while the run is open and has made @@ -1600,7 +1842,10 @@ export class Workspace extends DurableObject { .from(chatDrafts) .where(eq(chatDrafts.chatId, id)) .all(); + // Its build jobs end with it, and their rows go below. + this.#terminateJobs(this.#jobs.cancelRunning(id)); this.ctx.storage.transactionSync(() => { + this.#jobs.deleteChat(id); this.#db.delete(chatMessages).where(eq(chatMessages.chatId, id)).run(); this.#db.delete(chatSources).where(eq(chatSources.chatId, id)).run(); this.#db.delete(chatCodeRuns).where(eq(chatCodeRuns.chatId, id)).run(); diff --git a/apps/core/test/build-jobs.test.ts b/apps/core/test/build-jobs.test.ts new file mode 100644 index 00000000..08dc1ce2 --- /dev/null +++ b/apps/core/test/build-jobs.test.ts @@ -0,0 +1,1026 @@ +import { agentErrors } from "@grasp-os/shared/agent"; +import { appErrors } from "@grasp-os/shared/apps"; +import { chatIdSchema } from "@grasp-os/shared/ids"; +import { roleErrors } from "@grasp-os/shared/roles"; +import { runInDurableObject } from "cloudflare:test"; +import { env } from "cloudflare:workers"; +import { drizzle } from "drizzle-orm/durable-sqlite"; +import { describe, expect, it, vi } from "vite-plus/test"; +import { z } from "zod"; + +import { BuildJobRecords, jobRecordsKept } from "../src/build-job-records.ts"; +import { + chatOf, + codeResults, + codeStep, + model, + openTestRun, + pausedReply, + pointAtGateway, + says, + transcript, +} from "./agent-chat.ts"; +import type { WorkspaceStub } from "./agent-chat.ts"; +import { fakeGateway } from "./ai-gateway.ts"; +import type { GatewayReply } from "./ai-gateway.ts"; +import { release, requestGranted } from "./apps.ts"; +import { allEvents } from "./audit-events.ts"; +import { mockIdp } from "./idp.ts"; +import { collectionWithNote, readCollection } from "./knowledge.ts"; +import { signedInApi } from "./sign-in.ts"; + +// A build a chat's code hands to a durable job (`env.build.submit`): it +// goes on after the code run that submitted it ends, as a workflow of +// core's own, and how it ended reaches the turn that submitted it. These +// tests start from how that can go wrong: the result reaches a turn that +// has ended, or a later one; a permission revoked while the job runs +// doesn't stop what it does next; and a restart of the job makes its +// commit twice. + +const idp = mockIdp(); + +const appName = "Invoice desk"; + +/** A screen the lint lets through. */ +const desk = `import { Button } from "@grasp-os/ui/components/button"; + +export default function Desk() { + return ( +
+ +
+ ); +} +`; + +/** A workflow of the draft, with its tests: what a check keeps. */ +const intake = { + "workflows/intake.ts": `import { workflow, z } from "@grasp-os/sdk/workflow"; + +export default workflow("intake", { params: {}, input: z.unknown() }, async (step, { input }) => { + return await step.do("read", { description: "Read the invoice" }, async () => input); +}); +`, + "workflows/intake.workflow-tests.ts": `import { workflowTests } from "@grasp-os/sdk/testing"; + +import definition from "./intake.ts"; + +export default workflowTests(definition, [ + { name: "runs", mocks: { read: 1 }, expect: {} }, +]); +`, +}; + +/** The screen with its button restyled, which @shadcn/lint refuses. */ +const restyled = desk.replace( + '