zopu-code/packages/backend/convex/workPlanning.ts

import {
  type WorkDefinition,
  validateDefinition,
  WorkQuestion,
} from "@code/primitives/work-definition";
import { validateDesignPacket } from "@code/primitives/work-design";
import { makeFunctionReference } from "convex/server";
import { ConvexError, v } from "convex/values";
import { Effect, Schema } from "effect";

import type { Doc, Id } from "./_generated/dataModel";
import {
  env,
  internalAction,
  internalQuery,
  mutation,
  query,
} from "./_generated/server";
import { requireProjectMember } from "./authz";

const plannerRef = makeFunctionReference<
  "action",
  { organizationId: Id<"organizations">; workId: Id<"works"> }
>("workPlanning:runPlanner");
const plannerContextRef = makeFunctionReference<
  "query",
  { workId: Id<"works"> },
  { objective: string; status: string; title: string } | null
>("workPlanning:getPlannerContext");

const parsePayload = (payloadJson: string): unknown => {
  try {
    return JSON.parse(payloadJson) as unknown;
  } catch {
    throw new ConvexError("Proposal payload must be valid JSON");
  }
};

const objectPayload = (payloadJson: string): Record<string, unknown> => {
  const value = parsePayload(payloadJson);
  if (typeof value !== "object" || value === null || Array.isArray(value))
    throw new ConvexError("Proposal payload must be an object");
  return value as Record<string, unknown>;
};

const requireWork = async (ctx: { db: any }, workId: Id<"works">) => {
  const work = (await ctx.db.get(workId)) as Doc<"works"> | null;
  if (!work) throw new ConvexError("Work not found");
  return work;
};

const appendEvent = async (
  ctx: { db: any },
  workId: Id<"works">,
  kind: any,
  idempotencyKey: string,
  referenceId?: string,
  payloadJson?: string
) => {
  const existing = await ctx.db
    .query("workEvents")
    .withIndex("by_work_and_idempotencyKey", (q: any) =>
      q.eq("workId", workId).eq("idempotencyKey", idempotencyKey)
    )
    .unique();
  if (existing) return existing._id;
  return await ctx.db.insert("workEvents", {
    workId,
    kind,
    idempotencyKey,
    createdAt: Date.now(),
    ...(referenceId ? { referenceId } : {}),
    ...(payloadJson ? { payloadJson } : {}),
  });
};

const invalidateApprovalsAndDesign = async (
  ctx: { db: any },
  workId: Id<"works">
) => {
  const approvals = await ctx.db
    .query("workApprovals")
    .withIndex("by_work_and_kind", (q: any) => q.eq("workId", workId))
    .collect();
  for (const approval of approvals) {
    if (approval.status === "active")
      await ctx.db.patch(approval._id, { status: "invalidated" });
  }
  const designs = await ctx.db
    .query("designPackets")
    .withIndex("by_work_and_version", (q: any) => q.eq("workId", workId))
    .collect();
  for (const design of designs) {
    if (design.status === "current")
      await ctx.db.patch(design._id, { status: "superseded" });
  }
  const slices = await ctx.db
    .query("workSlices")
    .withIndex("by_work_and_designVersion", (q: any) => q.eq("workId", workId))
    .collect();
  for (const slice of slices) await ctx.db.delete(slice._id);
};

export const requestDefinition = mutation({
  args: { workId: v.id("works") },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    if (work.status !== "proposed" && work.status !== "defining")
      throw new ConvexError("Work is not available for definition");
    const now = Date.now();
    await ctx.db.patch(work._id, { status: "defining", updatedAt: now });
    await appendEvent(
      ctx,
      work._id,
      "definition.requested",
      `definition-requested:${work._id}`
    );
    await ctx.scheduler.runAfter(0, plannerRef, {
      organizationId: work.organizationId,
      workId: work._id,
    });
    return { status: "defining" as const };
  },
});

const saveDefinition = async (
  ctx: { db: any },
  work: Doc<"works">,
  payloadJson: string,
  createdBy: string
) => {
  if (work.status === "executing")
    throw new ConvexError("Cannot revise Work while a Run is executing");
  const decoded = await Effect.runPromise(
    validateDefinition({
      ...objectPayload(payloadJson),
      version: (work.definitionVersion ?? 0) + 1,
    })
  ).catch((error: unknown) => {
    throw new ConvexError(
      error instanceof Error ? error.message : "Invalid Definition"
    );
  });
  const version = decoded.version;
  await invalidateApprovalsAndDesign(ctx, work._id);
  const previous = await ctx.db
    .query("workDefinitions")
    .withIndex("by_work_and_version", (q: any) => q.eq("workId", work._id))
    .collect();
  for (const row of previous)
    if (row.status === "current")
      await ctx.db.patch(row._id, { status: "superseded" });
  const definitionId = await ctx.db.insert("workDefinitions", {
    workId: work._id,
    version,
    payloadJson: JSON.stringify(decoded),
    risk: decoded.risk,
    status: "current",
    createdBy,
    createdAt: Date.now(),
  });
  const questions = await ctx.db
    .query("workQuestions")
    .withIndex("by_work_and_definitionVersion", (q: any) =>
      q.eq("workId", work._id)
    )
    .collect();
  for (const question of questions)
    if (question.definitionVersion !== version)
      await ctx.db.delete(question._id);
  for (const question of decoded.questions) {
    await ctx.db.insert("workQuestions", {
      workId: work._id,
      definitionVersion: version,
      questionId: question.id,
      prompt: question.prompt,
      impact: question.impact,
      recommendation: question.recommendation,
      alternativesJson: JSON.stringify(question.alternatives),
      status: question.status,
      answer: question.answer,
      createdAt: Date.now(),
    });
  }
  await ctx.db.patch(work._id, {
    definitionVersion: version,
    definitionApprovalVersion: undefined,
    designVersion: undefined,
    designApprovalVersion: undefined,
    status: "awaiting-definition-approval",
    updatedAt: Date.now(),
  });
  await appendEvent(
    ctx,
    work._id,
    work.definitionVersion ? "definition.revised" : "definition.saved",
    `definition:${version}`,
    String(definitionId),
    JSON.stringify(decoded)
  );
  return { definitionId, version };
};

export const saveDefinitionProposal = mutation({
  args: { workId: v.id("works"), payloadJson: v.string() },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    return await saveDefinition(ctx, work, args.payloadJson, "user");
  },
});

export const reviseDefinition = saveDefinitionProposal;

export const approveDefinition = mutation({
  args: { workId: v.id("works"), version: v.number() },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    if (
      work.definitionVersion !== args.version ||
      work.status !== "awaiting-definition-approval"
    )
      throw new ConvexError("Definition version is not current or approvable");
    const row = await ctx.db
      .query("workDefinitions")
      .withIndex("by_work_and_version", (q: any) =>
        q.eq("workId", work._id).eq("version", args.version)
      )
      .unique();
    if (!row) throw new ConvexError("Definition not found");
    const definition = JSON.parse(row.payloadJson) as WorkDefinition;
    const valid = await Effect.runPromise(validateDefinition(definition)).catch(
      (error: unknown) => {
        throw new ConvexError(
          error instanceof Error
            ? error.message
            : "Definition cannot be approved"
        );
      }
    );
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new ConvexError("Authentication required");
    await ctx.db.insert("workApprovals", {
      workId: work._id,
      kind: "definition",
      definitionVersion: args.version,
      approvedBy: identity.subject,
      approvedAt: Date.now(),
      status: "active",
    });
    await ctx.db.patch(work._id, {
      status: "designing",
      definitionApprovalVersion: valid.version,
      updatedAt: Date.now(),
    });
    await appendEvent(
      ctx,
      work._id,
      "definition.approved",
      `definition-approved:${args.version}`
    );
    await ctx.scheduler.runAfter(0, plannerRef, {
      organizationId: work.organizationId,
      workId: work._id,
    });
    return { status: "designing" as const, version: args.version };
  },
});

const reviseQuestion = async (
  ctx: { db: any },
  work: Doc<"works">,
  questionId: string,
  update: { status: "answered" | "withdrawn"; answer?: string }
) => {
  const current = await ctx.db
    .query("workDefinitions")
    .withIndex("by_work_and_version", (q: any) =>
      q.eq("workId", work._id).eq("version", work.definitionVersion ?? 0)
    )
    .unique();
  if (!current) throw new ConvexError("Current Definition not found");
  const definition = JSON.parse(current.payloadJson) as WorkDefinition;
  const next = {
    ...definition,
    questions: definition.questions.map((question) =>
      question.id === questionId ? { ...question, ...update } : question
    ),
  };
  return await saveDefinition(ctx, work, JSON.stringify(next), "user");
};

export const answerQuestion = mutation({
  args: { workId: v.id("works"), questionId: v.string(), answer: v.string() },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    const result = await reviseQuestion(ctx, work, args.questionId, {
      status: "answered",
      answer: args.answer,
    });
    await appendEvent(
      ctx,
      work._id,
      "question.answered",
      `question-answered:${args.questionId}:${result.version}`
    );
    return result;
  },
});

export const withdrawQuestion = mutation({
  args: { workId: v.id("works"), questionId: v.string() },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    const result = await reviseQuestion(ctx, work, args.questionId, {
      status: "withdrawn",
    });
    await appendEvent(
      ctx,
      work._id,
      "question.withdrawn",
      `question-withdrawn:${args.questionId}:${result.version}`
    );
    return result;
  },
});

export const requestDesign = mutation({
  args: { workId: v.id("works") },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    if (
      work.status !== "designing" ||
      work.definitionApprovalVersion !== work.definitionVersion
    )
      throw new ConvexError("Definition must be approved before Design");
    await appendEvent(
      ctx,
      work._id,
      "design.requested",
      `design-requested:${work.definitionVersion}`
    );
    return { status: work.status };
  },
});

const saveDesign = async (
  ctx: { db: any },
  work: Doc<"works">,
  payloadJson: string,
  createdBy: string
) => {
  if (work.status === "executing")
    throw new ConvexError("Cannot revise Work while a Run is executing");
  if (work.definitionApprovalVersion !== work.definitionVersion)
    throw new ConvexError("Design must bind an approved Definition version");
  const decoded = await Effect.runPromise(
    validateDesignPacket({
      ...objectPayload(payloadJson),
      version: (work.designVersion ?? 0) + 1,
      definitionVersion: work.definitionVersion,
    })
  ).catch((error: unknown) => {
    throw new ConvexError(
      error instanceof Error ? error.message : "Invalid Design Packet"
    );
  });
  const prior = await ctx.db
    .query("designPackets")
    .withIndex("by_work_and_version", (q: any) => q.eq("workId", work._id))
    .collect();
  for (const row of prior)
    if (row.status === "current")
      await ctx.db.patch(row._id, { status: "superseded" });
  const designApprovals = await ctx.db
    .query("workApprovals")
    .withIndex("by_work_and_kind", (q: any) =>
      q.eq("workId", work._id).eq("kind", "design")
    )
    .collect();
  for (const approval of designApprovals)
    if (approval.status === "active")
      await ctx.db.patch(approval._id, { status: "invalidated" });
  const designId = await ctx.db.insert("designPackets", {
    workId: work._id,
    version: decoded.version,
    definitionVersion: decoded.definitionVersion,
    payloadJson: JSON.stringify(decoded),
    status: "current",
    createdBy,
    createdAt: Date.now(),
  });
  const priorSlices = await ctx.db
    .query("workSlices")
    .withIndex("by_work_and_designVersion", (q: any) =>
      q.eq("workId", work._id)
    )
    .collect();
  for (const slice of priorSlices) await ctx.db.delete(slice._id);
  for (const [ordinal, slice] of decoded.slices.entries())
    await ctx.db.insert("workSlices", {
      workId: work._id,
      designVersion: decoded.version,
      sliceId: slice.id,
      ordinal,
      title: slice.title,
      objective: slice.objective,
      observableBehavior: slice.observableBehavior,
      payloadJson: JSON.stringify(slice),
      status: ordinal === 0 ? "ready" : "planned",
    });
  await ctx.db.patch(work._id, {
    designVersion: decoded.version,
    designApprovalVersion: undefined,
    status: "awaiting-design-approval",
    updatedAt: Date.now(),
  });
  await appendEvent(
    ctx,
    work._id,
    work.designVersion ? "design.revised" : "design.saved",
    `design:${decoded.version}`,
    String(designId),
    JSON.stringify(decoded)
  );
  return { designId, version: decoded.version };
};

export const saveDesignProposal = mutation({
  args: { workId: v.id("works"), payloadJson: v.string() },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    return await saveDesign(ctx, work, args.payloadJson, "user");
  },
});

export const reviseDesign = saveDesignProposal;

export const approveDesign = mutation({
  args: {
    workId: v.id("works"),
    definitionVersion: v.number(),
    designVersion: v.number(),
  },
  handler: async (ctx, args) => {
    const work = await requireWork(ctx, args.workId);
    await requireProjectMember(ctx, work.projectId);
    if (
      work.definitionApprovalVersion !== args.definitionVersion ||
      work.designVersion !== args.designVersion ||
      work.status !== "awaiting-design-approval"
    )
      throw new ConvexError(
        "Design approval must bind current Definition and Design versions"
      );
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new ConvexError("Authentication required");
    await ctx.db.insert("workApprovals", {
      workId: work._id,
      kind: "design",
      definitionVersion: args.definitionVersion,
      designVersion: args.designVersion,
      approvedBy: identity.subject,
      approvedAt: Date.now(),
      status: "active",
    });
    await ctx.db.patch(work._id, {
      status: "ready",
      designApprovalVersion: args.designVersion,
      updatedAt: Date.now(),
    });
    await appendEvent(
      ctx,
      work._id,
      "design.approved",
      `design-approved:${args.definitionVersion}:${args.designVersion}`
    );
    return { status: "ready" as const };
  },
});

export const submitDefinitionProposal = mutation({
  args: { workId: v.id("works"), payloadJson: v.string(), token: v.string() },
  handler: async (ctx, args) => {
    if (args.token !== env.FLUE_DB_TOKEN)
      throw new ConvexError("Invalid agent control token");
    const work = await requireWork(ctx, args.workId);
    return await saveDefinition(ctx, work, args.payloadJson, "work-planner");
  },
});

export const submitDesignProposal = mutation({
  args: { workId: v.id("works"), payloadJson: v.string(), token: v.string() },
  handler: async (ctx, args) => {
    if (args.token !== env.FLUE_DB_TOKEN)
      throw new ConvexError("Invalid agent control token");
    const work = await requireWork(ctx, args.workId);
    return await saveDesign(ctx, work, args.payloadJson, "work-planner");
  },
});

export const submitQuestion = mutation({
  args: { workId: v.id("works"), questionJson: v.string(), token: v.string() },
  handler: async (ctx, args) => {
    if (args.token !== env.FLUE_DB_TOKEN)
      throw new ConvexError("Invalid agent control token");
    const work = await requireWork(ctx, args.workId);
    if (work.definitionVersion === undefined)
      throw new ConvexError("Work has no Definition to attach a question to");
    const question = await Effect.runPromise(
      Schema.decodeUnknownEffect(WorkQuestion)(JSON.parse(args.questionJson))
    ).catch((error: unknown) => {
      throw new ConvexError(
        error instanceof Error ? error.message : "Invalid question"
      );
    });
    await ctx.db.insert("workQuestions", {
      workId: work._id,
      definitionVersion: work.definitionVersion,
      questionId: question.id,
      prompt: question.prompt,
      impact: question.impact,
      recommendation: question.recommendation,
      alternativesJson: JSON.stringify(question.alternatives),
      status: question.status,
      answer: question.answer,
      createdAt: Date.now(),
    });
    return { accepted: true };
  },
});

export const runPlanner = internalAction({
  args: { organizationId: v.id("organizations"), workId: v.id("works") },
  handler: async (ctx, args) => {
    const flueUrl = (env as typeof env & { readonly FLUE_URL?: string })
      .FLUE_URL;
    if (!flueUrl) return null;
    const context = await ctx.runQuery(plannerContextRef, {
      workId: args.workId,
    });
    if (!context) return null;
    const endpoint = new URL(
      `agents/work-planner/${encodeURIComponent(String(args.organizationId))}`,
      `${flueUrl.replace(/\/+$/u, "")}/`
    );
    endpoint.searchParams.set("wait", "result");
    await fetch(endpoint, {
      method: "POST",
      headers: {
        authorization: `Bearer ${env.FLUE_DB_TOKEN}`,
        "content-type": "application/json",
        "x-zopu-organization-id": String(args.organizationId),
        "x-zopu-request-id": `work-planner:${args.workId}`,
      },
      body: JSON.stringify({
        message: `Plan Work ${String(args.workId)} titled "${context.title}" with objective "${context.objective}". Current state is ${context.status}. If the Work is defining, submit only a Definition proposal and questions. If the Work is designing, submit only a Design Packet proposal bound to the approved Definition. Never approve or execute.`,
      }),
    });
    // The private worker submits typed proposals through Convex mutations; it
    // never approves or advances Work itself.
    return null;
  },
});

export const getPlannerContext = internalQuery({
  args: { workId: v.id("works") },
  handler: async (ctx, args) => {
    const work = await ctx.db.get(args.workId);
    return work
      ? { objective: work.objective, status: work.status, title: work.title }
      : null;
  },
});

export const listForProject = query({
  args: { projectId: v.id("projects") },
  handler: async (ctx, args) => {
    await requireProjectMember(ctx, args.projectId);
    const works = await ctx.db
      .query("works")
      .withIndex("by_project_and_createdAt", (q: any) =>
        q.eq("projectId", args.projectId)
      )
      .order("desc")
      .take(100);
    return await Promise.all(
      works.map(async (work) => {
        const definitions = await ctx.db
          .query("workDefinitions")
          .withIndex("by_work_and_version", (q: any) =>
            q.eq("workId", work._id)
          )
          .order("desc")
          .take(10);
        const designs = await ctx.db
          .query("designPackets")
          .withIndex("by_work_and_version", (q: any) =>
            q.eq("workId", work._id)
          )
          .order("desc")
          .take(10);
        const slices = await ctx.db
          .query("workSlices")
          .withIndex("by_work_and_designVersion", (q: any) =>
            q.eq("workId", work._id)
          )
          .collect();
        const runs = await ctx.db
          .query("workRuns")
          .withIndex("by_work_and_createdAt", (q: any) =>
            q.eq("workId", work._id)
          )
          .order("desc")
          .take(10);
        const events = await ctx.db
          .query("workEvents")
          .withIndex("by_work_and_createdAt", (q: any) =>
            q.eq("workId", work._id)
          )
          .order("desc")
          .take(100);
        const attachments = await ctx.db
          .query("signalWorkAttachments")
          .withIndex("by_work", (q: any) => q.eq("workId", work._id))
          .collect();
        const signals = await Promise.all(
          attachments.map(async (attachment) => {
            const signal = await ctx.db.get(attachment.signalId);
            if (!signal) return null;
            const sources = await ctx.db
              .query("signalSources")
              .withIndex("by_signalId_and_ordinal", (q: any) =>
                q.eq("signalId", signal._id)
              )
              .collect();
            return {
              createdAt: signal.createdAt,
              signalId: signal._id,
              summary: signal.summary,
              title: signal.title,
              sources: sources
                .sort((a, b) => a.ordinal - b.ordinal)
                .map((source) => ({
                  createdAt: source.sourceCreatedAt,
                  messageId: String(source.messageId),
                  rawText: source.rawTextSnapshot,
                  submissionId: null,
                })),
            };
          })
        );
        const currentDefinition = definitions.find(
          (definition) => definition.status === "current"
        );
        const currentDesign = designs.find(
          (design) => design.status === "current"
        );
        return {
          ...work,
          definitions,
          designs,
          slices: slices.sort((a, b) => a.ordinal - b.ordinal),
          runs,
          events,
          signals: signals.filter((signal) => signal !== null),
          definition: currentDefinition
            ? JSON.parse(currentDefinition.payloadJson)
            : null,
          design: currentDesign ? JSON.parse(currentDesign.payloadJson) : null,
        };
      })
    );
  },
});