feat(agy): queue turns, stream output, expose usage and image attachments

apps/server/src/persistence/Layers/ProjectionTurns.ts

@@ -10,6 +10,7 @@ import * as Struct from "effect/Struct";
 import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts";
 import {
   ClearCheckpointTurnConflictInput,
+  DeleteProjectionPendingTurnStartByMessageIdInput,
   DeleteProjectionTurnsByThreadInput,
   GetProjectionPendingTurnStartInput,
   GetProjectionTurnByTurnIdInput,
@@ -96,7 +97,7 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
       `,
   });
 
-  const clearPendingProjectionTurnsByThread = SqlSchema.void({
+  const clearOrConsumePendingProjectionTurnsByThread = SqlSchema.void({
     Request: DeleteProjectionTurnsByThreadInput,
     execute: ({ threadId }) =>
       sql`
@@ -105,6 +106,23 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
           AND turn_id IS NULL
           AND state = 'pending'
           AND checkpoint_turn_count IS NULL
+          AND (
+            COALESCE(
+              (SELECT status FROM projection_thread_sessions WHERE thread_id = ${threadId}),
+              ''
+            ) <> 'running'
+            OR row_id = (
+              SELECT queued.row_id
+              FROM projection_turns AS queued
+              WHERE queued.thread_id = ${threadId}
+                AND queued.turn_id IS NULL
+                AND queued.state = 'pending'
+                AND queued.pending_message_id IS NOT NULL
+                AND queued.checkpoint_turn_count IS NULL
+              ORDER BY queued.requested_at ASC, queued.row_id ASC
+              LIMIT 1
+            )
+          )
       `,
   });
 
@@ -164,11 +182,24 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
           AND state = 'pending'
           AND pending_message_id IS NOT NULL
           AND checkpoint_turn_count IS NULL
-        ORDER BY requested_at DESC
+        ORDER BY requested_at ASC, row_id ASC
         LIMIT 1
       `,
   });
 
+  const deletePendingProjectionTurnByMessageId = SqlSchema.void({
+    Request: DeleteProjectionPendingTurnStartByMessageIdInput,
+    execute: ({ threadId, messageId }) =>
+      sql`
+        DELETE FROM projection_turns
+        WHERE thread_id = ${threadId}
+          AND turn_id IS NULL
+          AND state = 'pending'
+          AND pending_message_id = ${messageId}
+          AND checkpoint_turn_count IS NULL
+      `,
+  });
+
   const listProjectionTurnsByThread = SqlSchema.findAll({
     Request: ListProjectionTurnsByThreadInput,
     Result: ProjectionTurnDbRowSchema,
@@ -264,21 +295,18 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
       ),
     );
 
-  const replacePendingTurnStart: ProjectionTurnRepositoryShape["replacePendingTurnStart"] = (row) =>
-    sql
-      .withTransaction(
-        clearPendingProjectionTurnsByThread({ threadId: row.threadId }).pipe(
-          Effect.flatMap(() => insertPendingProjectionTurn(row)),
-        ),
-      )
-      .pipe(
-        Effect.mapError(
-          toPersistenceSqlOrDecodeError(
-            "ProjectionTurnRepository.replacePendingTurnStart:query",
-            "ProjectionTurnRepository.replacePendingTurnStart:encodeRequest",
-          ),
+  const enqueuePendingTurnStart: ProjectionTurnRepositoryShape["enqueuePendingTurnStart"] = (row) =>
+    insertPendingProjectionTurn(row).pipe(
+      Effect.mapError(
+        toPersistenceSqlOrDecodeError(
+          "ProjectionTurnRepository.enqueuePendingTurnStart:query",
+          "ProjectionTurnRepository.enqueuePendingTurnStart:encodeRequest",
         ),
-      );
+      ),
+    );
+
+  const replacePendingTurnStart: ProjectionTurnRepositoryShape["replacePendingTurnStart"] =
+    enqueuePendingTurnStart;
 
   const getPendingTurnStartByThreadId: ProjectionTurnRepositoryShape["getPendingTurnStartByThreadId"] =
     (input) =>
@@ -290,12 +318,22 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
 
   const deletePendingTurnStartByThreadId: ProjectionTurnRepositoryShape["deletePendingTurnStartByThreadId"] =
     (input) =>
-      clearPendingProjectionTurnsByThread(input).pipe(
+      clearOrConsumePendingProjectionTurnsByThread(input).pipe(
         Effect.mapError(
           toPersistenceSqlError("ProjectionTurnRepository.deletePendingTurnStartByThreadId:query"),
         ),
       );
 
+  const deletePendingTurnStartByMessageId: ProjectionTurnRepositoryShape["deletePendingTurnStartByMessageId"] =
+    (input) =>
+      deletePendingProjectionTurnByMessageId(input).pipe(
+        Effect.mapError(
+          toPersistenceSqlError(
+            "ProjectionTurnRepository.deletePendingTurnStartByMessageId:query",
+          ),
+        ),
+      );
+
   const listByThreadId: ProjectionTurnRepositoryShape["listByThreadId"] = (input) =>
     listProjectionTurnsByThread(input).pipe(
       Effect.mapError(
@@ -339,9 +377,11 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
 
   return {
     upsertByTurnId,
+    enqueuePendingTurnStart,
     replacePendingTurnStart,
     getPendingTurnStartByThreadId,
     deletePendingTurnStartByThreadId,
+    deletePendingTurnStartByMessageId,
     listByThreadId,
     getByTurnId,
     clearCheckpointTurnConflict,

apps/server/src/persistence/Services/ProjectionTurns.ts

@@ -94,6 +94,13 @@ export const GetProjectionPendingTurnStartInput = Schema.Struct({
 });
 export type GetProjectionPendingTurnStartInput = typeof GetProjectionPendingTurnStartInput.Type;
 
+export const DeleteProjectionPendingTurnStartByMessageIdInput = Schema.Struct({
+  threadId: ThreadId,
+  messageId: MessageId,
+});
+export type DeleteProjectionPendingTurnStartByMessageIdInput =
+  typeof DeleteProjectionPendingTurnStartByMessageIdInput.Type;
+
 export const DeleteProjectionTurnsByThreadInput = Schema.Struct({
   threadId: ThreadId,
 });
@@ -107,58 +114,52 @@ export const ClearCheckpointTurnConflictInput = Schema.Struct({
 export type ClearCheckpointTurnConflictInput = typeof ClearCheckpointTurnConflictInput.Type;
 
 export interface ProjectionTurnRepositoryShape {
-  /**
-   * Inserts or updates the canonical row for a concrete `{threadId, turnId}` turn lifecycle state.
-   */
   readonly upsertByTurnId: (
     row: ProjectionTurnById,
   ) => Effect.Effect<void, ProjectionRepositoryError>;
 
+  /** Appends a pending-start placeholder. Pending starts are consumed FIFO. */
+  readonly enqueuePendingTurnStart: (
+    row: ProjectionPendingTurnStart,
+  ) => Effect.Effect<void, ProjectionRepositoryError>;
+
   /**
-   * Replaces any existing pending-start placeholder rows for a thread with exactly one latest pending-start row.
+   * Compatibility entry point used by the existing projector. It now appends
+   * instead of replacing so multiple follow-up prompts can coexist.
    */
   readonly replacePendingTurnStart: (
     row: ProjectionPendingTurnStart,
   ) => Effect.Effect<void, ProjectionRepositoryError>;
 
-  /**
-   * Returns the newest pending-start placeholder for a thread; this is expected to be at most one row after replacement writes.
-   */
+  /** Returns the oldest pending-start placeholder for a thread. */
   readonly getPendingTurnStartByThreadId: (
     input: GetProjectionPendingTurnStartInput,
   ) => Effect.Effect<Option.Option<ProjectionPendingTurnStart>, ProjectionRepositoryError>;
 
   /**
-   * Deletes only pending-start placeholder rows (`turnId = null`) for a thread and leaves concrete turn rows untouched.
+   * Consumes the oldest pending start while the projected session is running;
+   * for terminal/non-running sessions it clears every pending start.
    */
   readonly deletePendingTurnStartByThreadId: (
     input: GetProjectionPendingTurnStartInput,
   ) => Effect.Effect<void, ProjectionRepositoryError>;
 
-  /**
-   * Lists all projection rows for a thread, including pending placeholders, with checkpoint rows ordered before non-checkpoint rows.
-   */
+  readonly deletePendingTurnStartByMessageId: (
+    input: DeleteProjectionPendingTurnStartByMessageIdInput,
+  ) => Effect.Effect<void, ProjectionRepositoryError>;
+
   readonly listByThreadId: (
     input: ListProjectionTurnsByThreadInput,
   ) => Effect.Effect<ReadonlyArray<ProjectionTurn>, ProjectionRepositoryError>;
 
-  /**
-   * Looks up a concrete turn row by `{threadId, turnId}` and never returns pending placeholder rows.
-   */
   readonly getByTurnId: (
     input: GetProjectionTurnByTurnIdInput,
   ) => Effect.Effect<Option.Option<ProjectionTurnById>, ProjectionRepositoryError>;
 
-  /**
-   * Clears checkpoint fields on conflicting rows that reuse the same checkpoint turn count in a thread, excluding the provided turn.
-   */
   readonly clearCheckpointTurnConflict: (
     input: ClearCheckpointTurnConflictInput,
   ) => Effect.Effect<void, ProjectionRepositoryError>;
 
-  /**
-   * Hard-deletes all projection rows for a thread, including pending-start placeholders and checkpoint metadata rows.
-   */
   readonly deleteByThreadId: (
     input: DeleteProjectionTurnsByThreadInput,
   ) => Effect.Effect<void, ProjectionRepositoryError>;

apps/server/src/provider/Drivers/AgyDriver.ts

@@ -3,9 +3,8 @@
  *
  * Wraps the `agy` binary in its documented headless mode (see
  * {@link ../Layers/AgyAdapter}) so each instance is one `agy` installation
- * addressed by `binaryPath`. No persistent process is owned by the driver —
- * one child per turn — so instances share nothing but the CLI's own
- * credential cache.
+ * addressed by `binaryPath`. The adapter owns one persistent stream-json
+ * process per live thread and resumes the conversation after a respawn.
  *
  * @module provider/Drivers/AgyDriver
  */
@@ -19,6 +18,7 @@ import { ChildProcessSpawner } from "effect/unstable/process";
 
 import { makeAgyTextGeneration } from "../../textGeneration/AgyTextGeneration.ts";
 import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts";
+import { ServerConfig } from "../../config.ts";
 import { ServerSettingsService } from "../../serverSettings.ts";
 import { ProviderDriverError } from "../Errors.ts";
 import { makeAgyAdapter } from "../Layers/AgyAdapter.ts";
@@ -60,6 +60,7 @@ export type AgyDriverEnv =
   | FileSystem.FileSystem
   | Path.Path
   | ProviderEventLoggers
+  | ServerConfig
   | ServerSettingsService;
 
 const withInstanceIdentity =

apps/server/src/provider/Layers/AgyAdapter.test.ts

@@ -1,6 +1,11 @@
 import { describe, expect, it } from "@effect/vitest";
 
-import { agyUserEventLine, parseAgyResume, usageFromAgy } from "./AgyAdapter.ts";
+import {
+  agyUserEventLine,
+  appendAgyImageAttachments,
+  parseAgyResume,
+  usageFromAgy,
+} from "./AgyAdapter.ts";
 
 describe("agyUserEventLine", () => {
   it("wraps a prompt as the NDJSON user event stream-json mode consumes", () => {
@@ -34,7 +39,7 @@ describe("parseAgyResume", () => {
 });
 
 describe("usageFromAgy", () => {
-  it("maps agy usage fields onto the canonical snapshot", () => {
+  it("adds cache reads back into canonical input/context usage", () => {
     const usage = usageFromAgy({
       input_tokens: 10415,
       output_tokens: 657,
@@ -43,16 +48,62 @@ describe("usageFromAgy", () => {
       total_tokens: 11072,
     }) as Record<string, unknown>;
     expect(usage).toBeDefined();
-    expect(usage.usedTokens).toBe(11072);
-    expect(usage.inputTokens).toBe(10415);
+    expect(usage.usedTokens).toBe(19185);
+    expect(usage.inputTokens).toBe(18528);
     expect(usage.outputTokens).toBe(657);
     expect(usage.reasoningOutputTokens).toBe(616);
     expect(usage.cachedInputTokens).toBe(8113);
+    expect(usage.lastInputTokens).toBe(18528);
+  });
+
+  it("turns cumulative persistent-session results into per-turn usage", () => {
+    const usage = usageFromAgy(
+      {
+        input_tokens: 30662,
+        output_tokens: 8,
+        thinking_tokens: 0,
+        cache_read_tokens: 30214,
+        total_tokens: 30670,
+      },
+      {
+        cumulativeResult: true,
+        previousCumulative: {
+          input_tokens: 30384,
+          output_tokens: 4,
+          thinking_tokens: 0,
+          cache_read_tokens: 0,
+          total_tokens: 30388,
+        },
+      },
+    ) as Record<string, unknown>;
+
+    expect(usage.usedTokens).toBe(30496);
+    expect(usage.inputTokens).toBe(30492);
+    expect(usage.cachedInputTokens).toBe(30214);
+    expect(usage.outputTokens).toBe(4);
+    expect(usage.totalProcessedTokens).toBe(30670);
   });
 
   it("returns undefined for non-object or non-numeric payloads", () => {
     expect(usageFromAgy(undefined)).toBeUndefined();
     expect(usageFromAgy("nope")).toBeUndefined();
-    expect(usageFromAgy({ total_tokens: "lots" })).toMatchObject({ usedTokens: 0 });
+    expect(usageFromAgy({ total_tokens: "lots" })).toBeUndefined();
+  });
+});
+
+describe("appendAgyImageAttachments", () => {
+  it("projects image metadata as a delimited path manifest", () => {
+    const prompt = appendAgyImageAttachments("Review this screenshot", [
+      {
+        name: "screen.png",
+        mimeType: "image/png",
+        sizeBytes: 1234,
+        path: "/tmp/t3/attachments/abc.png",
+      },
+    ]);
+    expect(prompt).toContain("Review this screenshot");
+    expect(prompt).toContain("<t3_attached_images>");
+    expect(prompt).toContain('"path":"/tmp/t3/attachments/abc.png"');
+    expect(prompt).toContain('"mimeType":"image/png"');
   });
 });