Reduce activity-write and observability overhead

.agents/references/scars/full.md

@@ -24,6 +24,8 @@
 - Runtime PID files published after HTTP startup are observations, not startup ownership. Hold a shared per-state-directory OS-backed claim before constructing runtime services/migrations, revalidate legacy runtime evidence under it, and never unlink the claim file to recover a crash.
 
 - Provider runtime activity is projected into orchestration domain events server-side before the web app consumes it.
+- Batch only contiguous same-thread tool lifecycle activities; persist events, projections, and receipts in one transaction, publish in order, and fall back to single-command processing on batch failure. Do not debounce provider errors, approvals, or turn/message completion boundaries.
+- Per-thread provider-log rotation does not bound aggregate disk use; apply age and total-byte retention across the provider log directory, and deterministically sample successful SQL spans while keeping failures.
 - Preserve command input on tool starts and normalize provider-specific nesting before compact activity projection; shell audit timing must identify native versus observed starts and never treat an unfinished provider call as OS-process liveness.
 - Runtime warning/error activities carry user-facing text in `payload.message` with optional `payload.detail`; every client's work-log derivation must surface both, or the rows render with nothing to expand.
 - Session startup/resume and turn lifecycle require predictable recovery: terminal reconciliation must settle the matching projected turn and clear `session.activeTurnId`; preserve a pre-acknowledgement start failure's `messageId`; and preserve terminal provider-event ordering during normal adapter shutdown.
@@ -258,6 +260,7 @@
 - A user-requested workflow run must be reconciled inline before its RPC resolves; leaving worker-thread creation to the coordinator's event-stream pass makes the client poll for a thread that does not exist yet. Keep the periodic sweep idempotent but cheap — never re-dispatch a worker turn for a thread that already has activity, or every domain event costs an artifact read plus a command-queue hop per incomplete run and delays the next requested run.
 - Once a new thread's first turn is accepted, route visibility and navigation failures must not trigger a compensating `thread.delete`; the provider may already be doing work that a lagging shell projection has not exposed yet.
 - Worktree cleanup intent must be persisted as a durable job from `thread.deleted`; suppress intent while any active thread owns a canonical path alias, canonicalize paths, reserve pending paths against reassignment, recheck active ownership under the worktree lock, release the lock before Git I/O while the pending row remains the reservation, bound retries for permanent failures, and never force-remove a dirty worktree. The durable cleanup worker must own provider and terminal teardown so restored or periodically swept jobs cannot bypass it; clients must not tear down runtime state before deletion commits.
+- Event-driven cleanup wakeups handle new work; periodic due-job scans are only restart/backlog recovery, so index and cap each scan instead of enqueueing an unbounded backlog against the shared database.
 - Durable cleanup intent and removal reservation are different lifecycles: waiting, dirty, or review-only rows must not block workspace admission, while cancellation and unarchive may not release a reservation once Git removal is in progress. A reserved removal must either complete or return through an explicit recovery path.
 - Acquire the removal reservation before provider/terminal teardown. Serialize unarchive and turn-start admission against that reservation; otherwise a stale queued cleanup can stop a newly resumed thread after eligibility has changed. Release the global lock before slow teardown and Git I/O.
 - Reconciliation must prove both repository identity and checkout association before trusting historical PR metadata. A stale PR number or a path that is merely present on disk is not authorization to remove a registered checkout.

apps/server/src/cli-config.test.ts

@@ -11,7 +11,7 @@ import { resolveServerConfig } from "./cli.ts";
 it.layer(NodeServices.layer)("cli config resolution", (it) => {
   const defaultObservabilityConfig = {
     traceMinLevel: "Info",
-    traceTimingEnabled: true,
+    traceTimingEnabled: false,
     traceBatchWindowMs: 200,
     traceMaxBytes: 10 * 1024 * 1024,
     traceMaxFiles: 10,

apps/server/src/cli.test.ts

@@ -2706,7 +2706,7 @@ it.layer(NodeServices.layer)("cli log-level parsing", (it) => {
             const cleanupJobs = yield* WorktreeCleanupJobRepository;
             const now = new Date(yield* Clock.currentTimeMillis).toISOString();
             const persistedJobs = yield* cleanupJobs.list();
-            const dueJobs = yield* cleanupJobs.listDue({ now });
+            const dueJobs = yield* cleanupJobs.listDue({ now, limit: 16 });
             assert.equal(dueJobs.length, 5, JSON.stringify({ now, persistedJobs }, null, 2));
             const reactor = yield* ThreadDeletionReactor;
             yield* reactor.start();

apps/server/src/cli.ts

@@ -293,7 +293,7 @@ const offlineFlag = Flag.boolean("offline").pipe(
 const EnvServerConfig = Config.all({
   logLevel: Config.logLevel("T3CODE_LOG_LEVEL").pipe(Config.withDefault("Info")),
   traceMinLevel: Config.logLevel("T3CODE_TRACE_MIN_LEVEL").pipe(Config.withDefault("Info")),
-  traceTimingEnabled: Config.boolean("T3CODE_TRACE_TIMING_ENABLED").pipe(Config.withDefault(true)),
+  traceTimingEnabled: Config.boolean("T3CODE_TRACE_TIMING_ENABLED").pipe(Config.withDefault(false)),
   traceFile: Config.string("T3CODE_TRACE_FILE").pipe(
     Config.option,
     Config.map(Option.getOrUndefined),

apps/server/src/cli/config.ts

@@ -42,7 +42,7 @@ export const resolveCliAuthConfig = (
     return {
       logLevel: Option.getOrElse(cliLogLevel, () => "Info"),
       traceMinLevel: "Info",
-      traceTimingEnabled: true,
+      traceTimingEnabled: false,
       traceBatchWindowMs: 200,
       traceMaxBytes: 10 * 1024 * 1024,
       traceMaxFiles: 10,

apps/server/src/config.ts

@@ -157,7 +157,7 @@ export class ServerConfig extends Context.Service<ServerConfig, ServerConfigShap
         return {
           logLevel: "Error",
           traceMinLevel: "Info",
-          traceTimingEnabled: true,
+          traceTimingEnabled: false,
           traceBatchWindowMs: 200,
           traceMaxBytes: 10 * 1024 * 1024,
           traceMaxFiles: 10,

apps/server/src/observability/Metrics.ts

@@ -31,6 +31,13 @@ export const orchestrationCommandAckDuration = Metric.timer(
   },
 );
 
+export const orchestrationActivityAppendBatchesTotal = Metric.counter(
+  "t3_orchestration_activity_append_batches_total",
+  {
+    description: "Number of committed transactions containing multiple activity append commands.",
+  },
+);
+
 export const orchestrationEventsProcessedTotal = Metric.counter(
   "t3_orchestration_events_processed_total",
   {

apps/server/src/observability/TraceSink.test.ts

@@ -5,10 +5,10 @@ import path from "node:path";
 import { assert, describe, it } from "@effect/vitest";
 import { Effect } from "effect";
 
-import type { TraceRecord } from "./TraceRecord.ts";
+import type { EffectTraceRecord, TraceRecord } from "./TraceRecord.ts";
 import { makeTraceSink } from "./TraceSink.ts";
 
-const makeRecord = (name: string, suffix = ""): TraceRecord => ({
+const makeRecord = (name: string, suffix = ""): EffectTraceRecord => ({
   type: "effect-span",
   name,
   traceId: `trace-${name}-${suffix}`,
@@ -63,6 +63,52 @@ describe("TraceSink", () => {
     ),
   );
 
+  it.effect("samples successful SQL spans while retaining SQL failures", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-trace-sql-sampling-"));
+        const tracePath = path.join(tempDir, "server.trace.ndjson");
+
+        try {
+          const sink = yield* makeTraceSink({
+            filePath: tracePath,
+            maxBytes: 1024 * 1024,
+            maxFiles: 2,
+            batchWindowMs: 10_000,
+          });
+
+          for (let index = 0; index < 100; index += 1) {
+            sink.push({
+              ...makeRecord("sql.query", String(index)),
+              attributes: { "db.system.name": "sqlite" },
+            });
+          }
+          sink.push({
+            ...makeRecord("sql.query.failure"),
+            attributes: { "db.system.name": "sqlite" },
+            exit: { _tag: "Failure", cause: "database unavailable" },
+          });
+          yield* sink.close();
+
+          const records = fs
+            .readFileSync(tracePath, "utf8")
+            .trim()
+            .split("\n")
+            .map((line) => JSON.parse(line) as TraceRecord);
+
+          assert.equal(
+            records.some((record) => record.name === "sql.query.failure"),
+            true,
+          );
+          assert.equal(records.length > 1, true);
+          assert.equal(records.length < 30, true);
+        } finally {
+          fs.rmSync(tempDir, { recursive: true, force: true });
+        }
+      }),
+    ),
+  );
+
   it.effect("rotates the trace file when the configured max size is exceeded", () =>
     Effect.scoped(
       Effect.gen(function* () {

apps/server/src/observability/TraceSink.ts

@@ -4,6 +4,29 @@ import { Effect } from "effect";
 import type { TraceRecord } from "./TraceRecord.ts";
 
 const FLUSH_BUFFER_THRESHOLD = 32;
+const SQL_TRACE_SAMPLE_RATE = 0.1;
+
+function shouldPersistTraceRecord(record: TraceRecord): boolean {
+  const isSqlSpan =
+    typeof record.attributes["db.system.name"] === "string" || /^sql\./i.test(record.name);
+  if (!isSqlSpan) return true;
+
+  const isFailure =
+    record.type === "effect-span"
+      ? record.exit._tag !== "Success"
+      : record.status?.code === "2" || record.status?.message !== undefined;
+  if (isFailure) return true;
+
+  // Trace/span ids are stable random identifiers, so deterministic head
+  // sampling retains a consistent 10% of successful SQL spans without
+  // coupling the decision to request ordering or adding mutable sampler state.
+  let hash = 2166136261;
+  for (const character of `${record.traceId}:${record.spanId}`) {
+    hash ^= character.charCodeAt(0);
+    hash = Math.imul(hash, 16777619);
+  }
+  return (hash >>> 0) / 0x1_0000_0000 < SQL_TRACE_SAMPLE_RATE;
+}
 
 export interface TraceSinkOptions {
   readonly filePath: string;
@@ -53,6 +76,7 @@ export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: Trac
   return {
     filePath: options.filePath,
     push(record) {
+      if (!shouldPersistTraceRecord(record)) return;
       try {
         buffer.push(`${JSON.stringify(record)}\n`);
         if (buffer.length >= FLUSH_BUFFER_THRESHOLD) {

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -1658,6 +1658,98 @@ describe("OrchestrationEngine", () => {
     await system.dispose();
   });
 
+  it("batches concurrent activity appends without changing their per-thread order", async () => {
+    const system = await createOrchestrationSystem();
+    const { engine } = system;
+    const createdAt = now();
+    const projectId = asProjectId("project-activity-batch");
+    const threadId = ThreadId.make("thread-activity-batch");
+
+    await system.run(
+      engine.dispatch({
+        type: "project.create",
+        commandId: CommandId.make("cmd-project-activity-batch"),
+        projectId,
+        title: "Activity batch",
+        workspaceRoot: "/tmp/project-activity-batch",
+        defaultModelSelection: null,
+        createdAt,
+      }),
+    );
+    await system.run(
+      engine.dispatch({
+        type: "thread.create",
+        commandId: CommandId.make("cmd-thread-activity-batch"),
+        threadId,
+        projectId,
+        title: "Activity batch",
+        modelSelection: {
+          instanceId: ProviderInstanceId.make("codex"),
+          model: "gpt-5-codex",
+        },
+        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+        runtimeMode: "full-access",
+        branch: null,
+        worktreePath: null,
+        createdAt,
+      }),
+    );
+
+    const commands = [0, 1, 2, 3].map((index) => ({
+      type: "thread.activity.append" as const,
+      commandId: CommandId.make(`cmd-activity-batch-${index}`),
+      threadId,
+      activity: {
+        id: EventId.make(`activity-batch-${index}`),
+        tone: "tool" as const,
+        kind: "tool.started",
+        summary: `Tool ${index} started`,
+        payload: { index },
+        turnId: TurnId.make("turn-activity-batch"),
+        createdAt,
+      },
+      createdAt,
+    }));
+
+    try {
+      const results = await system.run(
+        Effect.all(
+          commands.map((command) => engine.dispatch(command)),
+          {
+            concurrency: "unbounded",
+          },
+        ),
+      );
+      const events = await system.run(
+        Stream.runCollect(engine.readEvents(0)).pipe(
+          Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)),
+        ),
+      );
+      const activities = events.filter(
+        (event) => event.type === "thread.activity-appended" && event.payload.threadId === threadId,
+      );
+
+      expect(results.map((result) => result.sequence)).toEqual(
+        results.map((result) => result.sequence).toSorted((left, right) => left - right),
+      );
+      expect(
+        activities.map((event) =>
+          event.type === "thread.activity-appended" ? event.payload.activity.id : "",
+        ),
+      ).toEqual(commands.map((command) => command.activity.id));
+
+      const snapshots = await system.run(Metric.snapshot);
+      expect(
+        hasMetricSnapshot(snapshots, "t3_orchestration_activity_append_batches_total", {
+          aggregateKind: "thread",
+          batchSize: "4",
+        }),
+      ).toBe(true);
+    } finally {
+      await system.dispose();
+    }
+  });
+
   it("treats deleting an already-deleted queued turn as a no-op", async () => {
     const system = await createOrchestrationSystem();
     const { engine } = system;