Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
3 changes: 3 additions & 0 deletions apps/server/src/auth/RpcAuthorization.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,9 @@ describe("RPC authorization scopes", () => {
expect(requiredScopeForRpcMethod(WS_METHODS.agentSessionsImport)).toBe(
AuthOrchestrationOperateScope,
);
expect(requiredScopeForRpcMethod(WS_METHODS.agentSessionsList)).toBe(
AuthOrchestrationReadScope,
);
});

it("reads the reviewer menu under the same scope as the pull request it belongs to", () => {
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/auth/RpcAuthorization.ts
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,7 @@ export const RPC_REQUIRED_SCOPES = {
[WS_METHODS.filesystemBrowse]: AuthOrchestrationReadScope,
[WS_METHODS.agentSessionsScan]: AuthOrchestrationReadScope,
[WS_METHODS.agentSessionsImport]: AuthOrchestrationOperateScope,
[WS_METHODS.agentSessionsList]: AuthOrchestrationReadScope,
[WS_METHODS.assetsCreateUrl]: AuthOrchestrationReadScope,
[WS_METHODS.attachmentsCreateUploadUrl]: AuthOrchestrationOperateScope,
[WS_METHODS.attachmentsDelete]: AuthOrchestrationOperateScope,
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/environment/ServerEnvironment.ts
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,7 @@ export const make = Effect.gen(function* () {
threadPullRequestLinking: true,
environmentIcon: true,
projectCloneTracking: true,
agentSessionPicker: true,
...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
...(serverSelfUpdate === "boot-service" || desktopAppUpdate
? {
Expand Down
162 changes: 161 additions & 1 deletion apps/server/src/project/AgentSessionImporter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
type OrchestrationCommand,
type OrchestrationProjectShell,
type OrchestrationThread,
type OrchestrationThreadShell,
type ProviderSendTurnInput,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
Expand Down Expand Up @@ -59,7 +60,7 @@ import { TextGeneration } from "../textGeneration/TextGeneration.ts";
import { TerminalManager } from "../terminal/Manager.ts";
import { VcsStatusBroadcaster } from "../vcs/VcsStatusBroadcaster.ts";
import * as RepositoryIdentityResolver from "./RepositoryIdentityResolver.ts";
import { importRecentAgentThreads } from "./AgentSessionImporter.ts";
import { importRecentAgentThreads, listProjectAgentSessions } from "./AgentSessionImporter.ts";
import * as AgentSessionScanner from "./AgentSessionScanner.ts";

const PROJECT_ID = ProjectId.make("project-1");
Expand All @@ -79,6 +80,7 @@ const makeThread = (source: "codex" | "claudeAgent"): AgentSessionScanner.AgentS
{ role: "user", text: "Fix the bug", createdAt: "2026-08-24T10:00:00.000Z" },
{ role: "assistant", text: "Fixed", createdAt: "2026-08-24T10:01:00.000Z" },
],
messageCount: 2,
});

const makeThreadOutcome = (thread: AgentSessionScanner.AgentSessionThread) =>
Expand Down Expand Up @@ -198,7 +200,165 @@ const runImport = (input: {
Effect.provide(input.snapshots),
);

const makeRecordingEngine = (commands: Array<OrchestrationCommand>) =>
OrchestrationEngine.OrchestrationEngineService.of({
dispatch: (command) => Effect.sync(() => ({ sequence: commands.push(command) })),
readEvents: () => Stream.empty,
readThreadEvents: () => Stream.empty,
getThreadReplayStats: () => Effect.die("unused"),
streamDomainEvents: Stream.empty,
subscribeDomainEvents: Effect.succeed(Stream.empty),
latestSequence: Effect.succeed(0),
});

it.layer(NodeServices.layer)("AgentSessionImporter", (it) => {
describe("conversation picker", () => {
it.effect("lists sessions and links the ones a live thread already resumes", () =>
Effect.gen(function* () {
const recentThreadCalls: Array<unknown> = [];
const codex = makeThread("codex");
const claude = {
...makeThread("claudeAgent"),
title: "Claude title",
messages: [
{ role: "user" as const, text: "\n Tidy the README\nthen more", createdAt: "" },
],
messageCount: 240,
};
const scanner = AgentSessionScanner.AgentSessionScanner.of({
scan: Effect.die("unused"),
recentThreads: (_root, completed, options) => {
recentThreadCalls.push({ completed, options });
return Stream.make(
makeThreadOutcome(codex),
{ _tag: "Skipped" } as const,
makeThreadOutcome(claude),
);
},
});
const liveThreadId = ThreadId.make("thread-native-codex");
const deletedThreadId = ThreadId.make("thread-deleted-claude");
const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({
upsert: () => Effect.die("unused"),
getProvider: () => Effect.die("unused"),
recordImportedTranscript: () => Effect.die("unused"),
getBinding: () => Effect.die("unused"),
listThreadIds: () => Effect.die("unused"),
listBindings: () =>
Effect.succeed([
{
threadId: liveThreadId,
provider: ProviderDriverKind.make("codex"),
providerInstanceId: codex.providerInstanceId,
resumeCursor: { threadId: codex.providerSessionId },
lastSeenAt: "2026-08-24T10:00:00.000Z",
},
{
threadId: deletedThreadId,
provider: ProviderDriverKind.make("claudeAgent"),
providerInstanceId: claude.providerInstanceId,
resumeCursor: { threadId: deletedThreadId, resume: claude.providerSessionId },
lastSeenAt: "2026-08-24T10:00:00.000Z",
},
]),
});
const snapshots = Layer.mock(ProjectionSnapshotQuery.ProjectionSnapshotQuery)({
getProjectShellById: () => Effect.succeed(Option.some(makeProject())),
getThreadShellById: (threadId) =>
Effect.succeed(
threadId === liveThreadId
? Option.some({} as unknown as OrchestrationThreadShell)
: Option.none(),
),
});

const result = yield* listProjectAgentSessions({ projectId: PROJECT_ID }).pipe(
Effect.provideService(AgentSessionScanner.AgentSessionScanner, scanner),
Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, directory),
Effect.provide(snapshots),
);

expect(recentThreadCalls).toEqual([{ completed: [], options: { refresh: true } }]);
expect(result).toEqual({
truncated: false,
sessions: [
{
provider: "codex",
providerInstanceId: "codex",
providerSessionId: "codex-session",
title: "Imported codex thread",
preview: "Fix the bug",
messageCount: 2,
createdAt: codex.createdAt,
updatedAt: codex.updatedAt,
threadId: liveThreadId,
},
{
provider: "claudeAgent",
providerInstanceId: "claudeAgent",
providerSessionId: CLAUDE_SESSION_ID,
title: "Claude title",
preview: "Tidy the README",
messageCount: 240,
createdAt: claude.createdAt,
updatedAt: claude.updatedAt,
threadId: null,
},
],
});
}),
);

it.effect("imports one session from a fresh scan and returns its thread", () =>
Effect.gen(function* () {
const commands: Array<OrchestrationCommand> = [];
const bindings: Array<ProviderSessionDirectory.ProviderRuntimeBinding> = [];
const requested: Array<AgentSessionScanner.AgentSessionRecentThreadsOptions | undefined> =
[];
const session = {
providerInstanceId: ProviderInstanceId.make("claudeAgent"),
providerSessionId: CLAUDE_SESSION_ID,
};
const scanner = AgentSessionScanner.AgentSessionScanner.of({
scan: Effect.die("unused"),
recentThreads: (_root, _completed, options) => {
requested.push(options);
return Stream.succeed(makeThreadOutcome(makeThread("claudeAgent")));
},
});
const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({
upsert: (binding) => Effect.sync(() => void bindings.push(binding)),
getProvider: () => Effect.die("unused"),
recordImportedTranscript: () => Effect.void,
getBinding: () => Effect.succeedNone,
listThreadIds: () => Effect.die("unused"),
listBindings: () => Effect.die("unused"),
});

const result = yield* importRecentAgentThreads({ projectId: PROJECT_ID, session }).pipe(
Effect.provideService(AgentSessionScanner.AgentSessionScanner, scanner),
Effect.provideService(
OrchestrationEngine.OrchestrationEngineService,
makeRecordingEngine(commands),
),
Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, directory),
Effect.provide(makeSnapshotsLayer({ project: makeProject() })),
);

const threadId = `import:claudeAgent:${CLAUDE_SESSION_ID}`;
expect(requested).toEqual([{ session, refresh: true }]);
expect(result).toEqual({ importedCount: 1, skippedCount: 0, threadIds: [threadId] });
expect(commands.map((command) => command.type)).toEqual([
"thread.create",
"thread.history.import",
]);
expect(bindings).toMatchObject([
{ threadId, resumeCursor: { threadId, resume: CLAUDE_SESSION_ID } },
]);
}),
);
});

describe("importRecentAgentThreads", () => {
it.effect("uses the project root and stores provider-specific resume cursors", () =>
Effect.gen(function* () {
Expand Down
110 changes: 109 additions & 1 deletion apps/server/src/project/AgentSessionImporter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,16 @@ import {
ThreadId,
type AgentSessionImportInput,
type AgentSessionImportResult,
type AgentSessionListInput,
type AgentSessionListResult,
type AgentSessionSummary,
type OrchestrationThread,
} from "@t3tools/contracts";
import { normalizeProjectPathForComparison } from "@t3tools/shared/path";
import * as Crypto from "effect/Crypto";
import * as Effect from "effect/Effect";
import * as Option from "effect/Option";
import * as Result from "effect/Result";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";

Expand All @@ -29,6 +33,13 @@ import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSn
import * as ProviderSessionDirectory from "../provider/Services/ProviderSessionDirectory.ts";
import * as AgentSessionScanner from "./AgentSessionScanner.ts";

/**
* The picker reads each listed transcript in full to count its messages, so it
* stops at the newest sessions instead of reading the import budget's hundred.
*/
const MAX_LISTED_SESSIONS = 50;
const MAX_PREVIEW_CHARS = 200;

const CLAUDE_SESSION_ID_PATTERN =
/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;

Expand Down Expand Up @@ -132,6 +143,8 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu
const threads = scanner.recentThreads(
workspaceRoot,
completedSources.map((entry) => entry.source),
// The picker lists from a fresh scan, so import from one too.
input.session === undefined ? {} : { session: input.session, refresh: true },
);
const importedThreadIds = new Set<ThreadId>();
let importedCount = 0;
Expand Down Expand Up @@ -294,5 +307,100 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu
}),
);

return { importedCount, skippedCount } satisfies AgentSessionImportResult;
return {
importedCount,
skippedCount,
...(input.session === undefined ? {} : { threadIds: [...importedThreadIds] }),
} satisfies AgentSessionImportResult;
});

/** The provider session a binding resumes, in the shape each adapter writes its cursor. */
function boundSessionId(binding: ProviderSessionDirectory.ProviderRuntimeBinding): string | null {
const cursor = binding.resumeCursor;
if (typeof cursor !== "object" || cursor === null) return null;
const fields = cursor as Record<string, unknown>;
const id =
binding.provider === "codex"
? fields.threadId
: binding.provider === "claudeAgent"
? (fields.resume ?? fields.sessionId)
: undefined;
return typeof id === "string" && id.length > 0 ? id : null;
}

function previewText(thread: AgentSessionScanner.AgentSessionThread): string {
const prompt = thread.messages.find((message) => message.role === "user")?.text ?? "";
const firstLine = prompt
.split("\n")
.map((line) => line.trim())
.find((line) => line.length > 0);
return (firstLine ?? "").slice(0, MAX_PREVIEW_CHARS);
}

/**
* List the recent Claude Code and Codex conversations recorded for a project's
* directory, newest first, marking the ones a live thread already resumes.
*/
export const listProjectAgentSessions = Effect.fn("listProjectAgentSessions")(function* (
input: AgentSessionListInput,
) {
const scanner = yield* AgentSessionScanner.AgentSessionScanner;
const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
const readProjects = (cause: unknown) =>
new AgentSessionScanError({ operation: "read-projects", cause });
const project = yield* snapshots.getProjectShellById(input.projectId).pipe(
Effect.mapError(readProjects),
Effect.flatMap(
Option.match({
onNone: () =>
Effect.fail(new AgentSessionImportProjectNotFoundError({ projectId: input.projectId })),
onSome: Effect.succeed,
}),
),
);

const threads = yield* scanner.recentThreads(project.workspaceRoot, [], { refresh: true }).pipe(
Stream.filterMap((outcome) =>
outcome._tag === "Importable" ? Result.succeed(outcome.thread) : Result.failVoid,
),
Stream.take(MAX_LISTED_SESSIONS + 1),
Stream.runCollect,
);

const bindings = yield* directory.listBindings().pipe(Effect.mapError(readProjects));
const boundThreadIds = new Map<string, ThreadId>();
for (const binding of bindings) {
const sessionId = boundSessionId(binding);
if (sessionId === null || binding.providerInstanceId === undefined) continue;
boundThreadIds.set(`${binding.providerInstanceId}\0${sessionId}`, binding.threadId);
}

const sessions: Array<AgentSessionSummary> = [];
for (const thread of Array.from(threads).slice(0, MAX_LISTED_SESSIONS)) {
const boundThreadId = boundThreadIds.get(
`${thread.providerInstanceId}\0${thread.providerSessionId}`,
);
// Bindings outlive deleted threads; only a live thread is worth opening.
const liveThread =
boundThreadId === undefined
? Option.none()
: yield* snapshots.getThreadShellById(boundThreadId).pipe(Effect.mapError(readProjects));
sessions.push({
provider: thread.source,
providerInstanceId: thread.providerInstanceId,
providerSessionId: thread.providerSessionId,
title: thread.title,
preview: previewText(thread),
messageCount: thread.messageCount,
createdAt: thread.createdAt,
updatedAt: thread.updatedAt,
threadId: Option.isSome(liveThread) && boundThreadId !== undefined ? boundThreadId : null,
});
}

return {
sessions,
truncated: threads.length > MAX_LISTED_SESSIONS,
} satisfies AgentSessionListResult;
});
Loading