Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion apps/desktop/src/settings/DesktopClientSettings.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ import * as DesktopEnvironment from "../app/DesktopEnvironment.ts";
import * as DesktopClientSettings from "./DesktopClientSettings.ts";

const clientSettings: ClientSettings = {
autoOpenPlanSidebar: false,
confirmThreadArchive: true,
confirmThreadDelete: false,
dismissedProviderUpdateNotificationKeys: [],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ import { OrchestrationEngineLive } from "../src/orchestration/Layers/Orchestrati
import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts";
import * as ThreadBackgroundLiveness from "../src/orchestration/ThreadBackgroundLiveness.ts";
import * as ThreadPlanProgress from "../src/orchestration/ThreadPlanProgress.ts";
import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceiptBus.ts";
import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";
import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";
Expand Down Expand Up @@ -306,7 +307,10 @@ export const makeOrchestrationIntegrationHarness = (
checkpointStoreLayer,
providerLayer,
RuntimeReceiptBusTest,
).pipe(Layer.provideMerge(ThreadBackgroundLiveness.layer));
).pipe(
Layer.provideMerge(ThreadBackgroundLiveness.layer),
Layer.provideMerge(ThreadPlanProgress.layer),
);
const serverSettingsLayer = ServerSettingsService.layerTest();
const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
Layer.provideMerge(runtimeServicesLayer),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
import { RuntimeReceiptBusLive } from "./RuntimeReceiptBus.ts";
import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
Expand Down Expand Up @@ -296,6 +297,7 @@ describe("CheckpointReactor", () => {
const orchestrationLayer = OrchestrationEngineLive.pipe(
Layer.provide(OrchestrationProjectionSnapshotQueryLive),
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(OrchestrationProjectionPipelineLive),
Layer.provide(OrchestrationEventStoreLive),
Layer.provide(OrchestrationCommandReceiptRepositoryLive),
Expand All @@ -304,6 +306,7 @@ describe("CheckpointReactor", () => {
);
const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe(
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(RepositoryIdentityResolver.layer),
Layer.provide(SqlitePersistenceMemory),
);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
import {
OrchestrationProjectionPipeline,
Expand All @@ -57,6 +58,7 @@ async function createOrchestrationSystem() {
OrchestrationProjectionSnapshotQueryLive,
).pipe(
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(OrchestrationEventStoreLive),
Layer.provide(OrchestrationCommandReceiptRepositoryLive),
Layer.provide(RepositoryIdentityResolver.layer),
Expand Down Expand Up @@ -820,6 +822,7 @@ describe("OrchestrationEngine", () => {
OrchestrationEngineLive.pipe(
Layer.provide(OrchestrationProjectionSnapshotQueryLive),
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(OrchestrationProjectionPipelineLive),
Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),
Layer.provide(OrchestrationCommandReceiptRepositoryLive),
Expand Down Expand Up @@ -926,6 +929,7 @@ describe("OrchestrationEngine", () => {
OrchestrationEngineLive.pipe(
Layer.provide(OrchestrationProjectionSnapshotQueryLive),
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),
Layer.provide(OrchestrationEventStoreLive),
Layer.provide(OrchestrationCommandReceiptRepositoryLive),
Expand Down Expand Up @@ -1070,6 +1074,7 @@ describe("OrchestrationEngine", () => {
OrchestrationEngineLive.pipe(
Layer.provide(OrchestrationProjectionSnapshotQueryLive),
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),
Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)),
Layer.provide(OrchestrationCommandReceiptRepositoryLive),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import {
} from "./ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
import { ServerConfig } from "../../config.ts";
Expand Down Expand Up @@ -2666,6 +2667,7 @@ const engineLayer = it.layer(
OrchestrationEngineLive.pipe(
Layer.provide(OrchestrationProjectionSnapshotQueryLive),
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(OrchestrationProjectionPipelineLive),
Layer.provide(OrchestrationEventStoreLive),
Layer.provide(OrchestrationCommandReceiptRepositoryLive),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityRes
import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";

const asProjectId = (value: string): ProjectId => ProjectId.make(value);
Expand All @@ -29,6 +30,7 @@ const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.make(val
const projectionSnapshotLayer = it.layer(
OrchestrationProjectionSnapshotQueryLive.pipe(
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provideMerge(RepositoryIdentityResolver.layer),
Layer.provideMerge(SqlitePersistenceMemory),
Layer.provideMerge(NodeServices.layer),
Expand Down Expand Up @@ -446,6 +448,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
hasPendingUserInput: false,
hasActionableProposedPlan: false,
backgroundLiveness: null,
planProgress: null,
},
]);

Expand Down Expand Up @@ -1829,6 +1832,7 @@ it.effect(
const resolveCalls: string[] = [];
const layer = OrchestrationProjectionSnapshotQueryLive.pipe(
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provideMerge(
Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, {
resolve: (cwd: string) =>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ import {
} from "../../persistence/Errors.ts";
import { ProjectionCheckpoint } from "../../persistence/Services/ProjectionCheckpoints.ts";
import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts";
import { ThreadPlanProgressService } from "../ThreadPlanProgress.ts";
import { ProjectionProject } from "../../persistence/Services/ProjectionProjects.ts";
import { ProjectionState } from "../../persistence/Services/ProjectionState.ts";
import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";
Expand Down Expand Up @@ -310,6 +311,7 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st

const makeProjectionSnapshotQuery = Effect.gen(function* () {
const threadBackgroundLiveness = yield* ThreadBackgroundLivenessService;
const threadPlanProgress = yield* ThreadPlanProgressService;
const sql = yield* SqlClient.SqlClient;
const repositoryIdentityResolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
const repositoryIdentityResolutionConcurrency = 4;
Expand Down Expand Up @@ -1683,6 +1685,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness(
row.threadId,
),
planProgress: threadPlanProgress.getThreadPlanProgress(row.threadId),
} satisfies OrchestrationThreadShell)
: Result.failVoid,
),
Expand Down Expand Up @@ -1826,6 +1829,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness(
row.threadId,
),
planProgress: threadPlanProgress.getThreadPlanProgress(row.threadId),
}),
),
updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z",
Expand Down Expand Up @@ -2101,6 +2105,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness(
threadRow.value.threadId,
),
planProgress: threadPlanProgress.getThreadPlanProgress(threadRow.value.threadId),
} satisfies OrchestrationThreadShell);
});

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
import {
providerErrorLabel,
providerErrorLabelFromInstanceHint,
Expand Down Expand Up @@ -347,6 +348,7 @@ describe("ProviderCommandReactor", () => {
const orchestrationLayer = OrchestrationEngineLive.pipe(
Layer.provide(OrchestrationProjectionSnapshotQueryLive),
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(OrchestrationProjectionPipelineLive),
Layer.provide(OrchestrationEventStoreLive),
Layer.provide(OrchestrationCommandReceiptRepositoryLive),
Expand All @@ -355,6 +357,7 @@ describe("ProviderCommandReactor", () => {
);
const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe(
Layer.provide(ThreadBackgroundLiveness.layer),
Layer.provide(ThreadPlanProgress.layer),
Layer.provide(RepositoryIdentityResolver.layer),
Layer.provide(SqlitePersistenceMemory),
);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts";
import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";
Expand Down Expand Up @@ -241,6 +242,7 @@ describe("ProviderRuntimeIngestion", () => {
// Single shared liveness instance across ingestion (writer), the
// engine, and the snapshot query (reader).
Layer.provideMerge(ThreadBackgroundLiveness.layer),
Layer.provideMerge(ThreadPlanProgress.layer),
Layer.provideMerge(SqlitePersistenceMemory),
Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)),
Expand Down
10 changes: 10 additions & 0 deletions apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/Projectio
import { isGitRepository } from "../../git/Utils.ts";
import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts";
import { ThreadPlanProgressService } from "../ThreadPlanProgress.ts";
import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
import {
ProviderRuntimeIngestionService,
Expand Down Expand Up @@ -826,6 +827,7 @@ export function runtimeEventToActivities(

const make = Effect.gen(function* () {
const threadBackgroundLiveness = yield* ThreadBackgroundLivenessService;
const threadPlanProgress = yield* ThreadPlanProgressService;
const crypto = yield* Crypto.Crypto;
const orchestrationEngine = yield* OrchestrationEngineService;
const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
Expand Down Expand Up @@ -1894,6 +1896,14 @@ const make = Effect.gen(function* () {
yield* rememberTaskDescription(thread.id, event.payload.taskId, description);
}
}
// Working-indicator plan progress: current step while the turn runs,
// cleared on settle so a finished plan never lingers as stale UI.
if (event.type === "turn.plan.updated") {
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Outdated
threadPlanProgress.recordPlanProgress(thread.id, event.payload.plan);
} else if (event.type === "turn.completed" || event.type === "session.exited") {
threadPlanProgress.clearThreadPlanProgress(thread.id);
}
Comment thread
cursor[bot] marked this conversation as resolved.

// Sidebar background liveness: fed from the same lifecycle stream,
// read by the shell query at mapping time (no persistence).
switch (event.type) {
Expand Down
44 changes: 44 additions & 0 deletions apps/server/src/orchestration/ThreadPlanProgress.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
import { describe, expect, it } from "vite-plus/test";
import * as ThreadPlanProgress from "./ThreadPlanProgress.ts";

describe("ThreadPlanProgress", () => {
it("tracks the in-progress step and clears when the plan completes", () => {
const progress = ThreadPlanProgress.make();
const threadId = "t-plan-1";
progress.recordPlanProgress(threadId, [
{ step: "Audit failure paths", status: "completed" },
{ step: "Implement the fix", status: "inProgress" },
{ step: "Run targeted tests", status: "pending" },
]);
expect(progress.getThreadPlanProgress(threadId)).toEqual({
step: "Implement the fix",
completedSteps: 1,
totalSteps: 3,
});

progress.recordPlanProgress(threadId, [
{ step: "Audit failure paths", status: "completed" },
{ step: "Implement the fix", status: "completed" },
{ step: "Run targeted tests", status: "completed" },
]);
expect(progress.getThreadPlanProgress(threadId)).toBeNull();
});

it("falls back to the first non-completed step when nothing is in progress", () => {
const progress = ThreadPlanProgress.make();
const threadId = "t-plan-2";
progress.recordPlanProgress(threadId, [
{ step: "First", status: "pending" },
{ step: "Second", status: "pending" },
]);
expect(progress.getThreadPlanProgress(threadId)?.step).toBe("First");
});

it("clearThreadPlanProgress removes the entry (turn settled / session died)", () => {
const progress = ThreadPlanProgress.make();
const threadId = "t-plan-3";
progress.recordPlanProgress(threadId, [{ step: "Only step", status: "inProgress" }]);
progress.clearThreadPlanProgress(threadId);
expect(progress.getThreadPlanProgress(threadId)).toBeNull();
});
});
76 changes: 76 additions & 0 deletions apps/server/src/orchestration/ThreadPlanProgress.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
/**
* ThreadPlanProgressService - in-memory per-thread plan progress for the
* Working indicators (sidebar rows, in-chat working line).
*
* Plans are a progress annotation, not a surface of their own: the useful
* kernel of a turn.plan.updated event is "which step is the agent on right
* now". Ingestion records the current step here and the shell query reads it
* at mapping time — no persistence, no migration (same pattern as
* ThreadBackgroundLivenessService). Cleared when the turn settles or the
* session dies, so a finished plan never lingers as stale UI.
*
* @module ThreadPlanProgressService
*/
import * as Context from "effect/Context";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";

export interface ThreadPlanProgress {
readonly step: string;
readonly completedSteps: number;
readonly totalSteps: number;
}

interface PlanStepInput {
readonly step: string;
readonly status: string;
}

export class ThreadPlanProgressService extends Context.Service<
ThreadPlanProgressService,
{
/**
* Feed one turn.plan.updated payload. An all-completed plan clears the
* entry (the turn is wrapping up; nothing is "in progress" anymore).
*/
readonly recordPlanProgress: (threadId: string, plan: ReadonlyArray<PlanStepInput>) => void;

/** Turn settled or session died: the working indicator reverts to plain. */
readonly clearThreadPlanProgress: (threadId: string) => void;

readonly getThreadPlanProgress: (threadId: string) => ThreadPlanProgress | null;
}
>()("t3/orchestration/ThreadPlanProgress/ThreadPlanProgressService") {}

export function make(): ThreadPlanProgressService["Service"] {
const progressByThreadId = new Map<string, ThreadPlanProgress>();

return {
recordPlanProgress: (threadId, plan) => {
const totalSteps = plan.length;
const completedSteps = plan.filter((step) => step.status === "completed").length;
// Current step: the in-progress one, else the first pending one (a
// plan that was just written has no in-progress step yet).
const current =
plan.find((step) => step.status === "inProgress") ??
plan.find((step) => step.status !== "completed");
if (totalSteps === 0 || completedSteps === totalSteps || current === undefined) {
progressByThreadId.delete(threadId);
return;
}
progressByThreadId.set(threadId, {
step: current.step,
completedSteps,
totalSteps,
});
},

clearThreadPlanProgress: (threadId) => {
progressByThreadId.delete(threadId);
},

getThreadPlanProgress: (threadId) => progressByThreadId.get(threadId) ?? null,
};
}

export const layer = Layer.effect(ThreadPlanProgressService, Effect.sync(make));
Loading
Loading