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
4 changes: 4 additions & 0 deletions apps/server/src/auth/RpcAuthorization.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,10 @@ describe("RPC authorization scopes", () => {
expect(new Set(Object.keys(RPC_REQUIRED_SCOPES))).toEqual(new Set(WsRpcGroup.requests.keys()));
});

it("keeps session listing, which reads transcripts outside T3, behind filesystem read", () => {
expect(requiredScopeForRpcMethod(WS_METHODS.agentSessionsList)).toBe(AuthFilesystemReadScope);
});

it("keeps webhook delivery logs, which hold request bodies, behind operate scope", () => {
for (const method of [
WS_METHODS.scheduledTasksListWebhookDeliveries,
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 @@ -152,6 +152,7 @@ export const RPC_REQUIRED_SCOPES = {
[WS_METHODS.filesystemBrowse]: AuthFilesystemReadScope,
[WS_METHODS.filesystemGetMetadata]: AuthFilesystemReadScope,
[WS_METHODS.agentSessionsScan]: AuthOrchestrationReadScope,
[WS_METHODS.agentSessionsList]: AuthFilesystemReadScope,
[WS_METHODS.agentSessionsImport]: AuthOrchestrationOperateScope,
[WS_METHODS.assetsCreateUrl]: AuthOrchestrationReadScope,
[WS_METHODS.assetsPersistChatAttachments]: AuthOrchestrationOperateScope,
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/observability/RpcInstrumentation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,7 @@ const RPC_AGGREGATES = {
[WS_METHODS.filesystemBrowse]: "workspace",
[WS_METHODS.filesystemGetMetadata]: "workspace",
[WS_METHODS.agentSessionsScan]: "workspace",
[WS_METHODS.agentSessionsList]: "workspace",
[WS_METHODS.agentSessionsImport]: "workspace",
[WS_METHODS.assetsCreateUrl]: "workspace",
[WS_METHODS.assetsPersistChatAttachments]: "orchestration",
Expand Down
76 changes: 68 additions & 8 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2378,22 +2378,35 @@ describe("ClaudeAdapterV2 MCP credential channel", () => {
});

describe("ClaudeAdapterV2 native session identity", () => {
const openTurnWithOrdinal = (providerTurnOrdinal: number, nativeThreadHasTurns?: boolean) =>
const openTurnWithOrdinal = (
providerTurnOrdinal: number,
nativeThreadHasTurns?: boolean,
importedNativeThreadId?: string,
) =>
Effect.scoped(
Effect.gen(function* () {
const fileSystem = yield* FileSystem.FileSystem;
const idAllocator = yield* IdAllocator.IdAllocatorV2;
const path = yield* Path.Path;
const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({
prefix: "t3-claude-v2-session-identity-",
});
if (importedNativeThreadId !== undefined) {
const projectDir = path.join(attachmentsDir, "projects", "-workspace");
yield* fileSystem.makeDirectory(projectDir, { recursive: true });
yield* fileSystem.writeFileString(
path.join(projectDir, `${importedNativeThreadId}.jsonl`),
`{"type":"user","message":{"role":"user","content":"hello"},"sessionId":"${importedNativeThreadId}"}\n`,
);
}
const openedQueries: Array<ClaudeAdapterV2.ClaudeAgentSdkQueryOpenInput> = [];
const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({
instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID,
settings: DEFAULT_CLAUDE_SETTINGS,
environment: {},
environment: { CLAUDE_CONFIG_DIR: attachmentsDir },
attachmentsDir,
fileSystem,
path: yield* Path.Path,
path,
crypto: yield* Crypto.Crypto,
idAllocator,
queryRunner: {
Expand Down Expand Up @@ -2426,12 +2439,50 @@ describe("ClaudeAdapterV2 native session identity", () => {
modelSelection: CLAUDE_TEST_MODEL_SELECTION,
runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY,
});
const providerThread = yield* runtime.ensureThread({
threadId,
modelSelection: CLAUDE_TEST_MODEL_SELECTION,
runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY,
});
const now = yield* DateTime.now;
// Mirrors the provider thread a session import records: a strong native
// ref with no conversation head and no provider turns.
const providerThread =
importedNativeThreadId === undefined
? yield* runtime.ensureThread({
threadId,
modelSelection: CLAUDE_TEST_MODEL_SELECTION,
runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY,
})
: yield* runtime.resumeThread({
threadId,
modelSelection: CLAUDE_TEST_MODEL_SELECTION,
runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY,
providerThread: {
id: idAllocator.derive.providerThread({
driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
providerInstanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID,
nativeThreadId: importedNativeThreadId,
}),
driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
providerInstanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID,
providerSessionId: null,
appThreadId: threadId,
ownerNodeId: null,
nativeThreadRef: {
driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
nativeId: importedNativeThreadId,
strength: "strong",
},
nativeConversationHeadRef: null,
status: "not_loaded",
firstRunOrdinal: null,
lastRunOrdinal: null,
handoffIds: [],
forkedFrom: null,
pendingBackgroundTasks: [],
contextUsage: null,
nativeMetadata: null,
goal: null,
createdAt: now,
updatedAt: now,
},
});
yield* runtime.startTurn(
makeClaudeTestTurnInput({
threadId,
Expand Down Expand Up @@ -2470,6 +2521,15 @@ describe("ClaudeAdapterV2 native session identity", () => {
assert.equal(openedQueries[0]?.options.resume, undefined);
}),
);

it.effect("resumes an imported native session on its first T3 turn", () =>
Effect.gen(function* () {
const openedQueries = yield* openTurnWithOrdinal(1, false, "imported-native-session");
assert.equal(openedQueries.length, 1);
assert.equal(openedQueries[0]?.options.resume, "imported-native-session");
assert.equal(openedQueries[0]?.options.sessionId, undefined);
}),
);
});

const encodeJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
Expand Down
95 changes: 52 additions & 43 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10563,41 +10563,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
yield* Effect.addFinalizer(() => closeSession());

// Metadata-only transcripts cannot be resumed or recreated; fail resumeThread to trigger a fresh session and handoff.
const hasEmptyClaudeTranscript = Effect.fn("ClaudeAdapterV2.hasEmptyClaudeTranscript")(
function* (nativeThreadId: string) {
const configuredDir = adapterOptions.environment.CLAUDE_CONFIG_DIR?.trim() ?? "";
const projectsDir = path.join(
configuredDir.length > 0
? path.resolve(session.cwd, configuredDir)
: path.join(yield* HostProcess.HomeDirectory, ".claude"),
"projects",
);
// Any message in any transcript copy means Claude may still resume it.
let found = false;
for (const project of yield* fileSystem.readDirectory(projectsDir)) {
const transcriptPath = path.join(projectsDir, project, `${nativeThreadId}.jsonl`);
// Stray files beside the project dirs fail with ENOTDIR.
if (
!(yield* fileSystem.exists(transcriptPath).pipe(Effect.orElseSucceed(() => false)))
) {
continue;
}
const conversationLine = yield* fileSystem
.stream(transcriptPath)
.pipe(
Stream.decodeText,
Stream.splitLines,
Stream.filter(isClaudeConversationLine),
Stream.runHead,
);
if (Option.isSome(conversationLine)) {
return false;
}
found = true;
const claudeTranscriptState = Effect.fn("ClaudeAdapterV2.claudeTranscriptState")(function* (
nativeThreadId: string,
) {
const configuredDir = adapterOptions.environment.CLAUDE_CONFIG_DIR?.trim() ?? "";
const projectsDir = path.join(
configuredDir.length > 0
? path.resolve(session.cwd, configuredDir)
: path.join(yield* HostProcess.HomeDirectory, ".claude"),
"projects",
);
// Any message in any transcript copy means Claude may still resume it.
let found = false;
for (const project of yield* fileSystem.readDirectory(projectsDir)) {
const transcriptPath = path.join(projectsDir, project, `${nativeThreadId}.jsonl`);
// Stray files beside the project dirs fail with ENOTDIR.
if (
!(yield* fileSystem.exists(transcriptPath).pipe(Effect.orElseSucceed(() => false)))
) {
continue;
}
return found;
},
);
const conversationLine = yield* fileSystem
.stream(transcriptPath)
.pipe(
Stream.decodeText,
Stream.splitLines,
Stream.filter(isClaudeConversationLine),
Stream.runHead,
);
if (Option.isSome(conversationLine)) {
return "conversation" as const;
}
found = true;
}
return found ? ("empty" as const) : ("missing" as const);
});

const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = {
instanceId: adapterOptions.instanceId,
Expand Down Expand Up @@ -10689,16 +10689,25 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
nativeThreadId !== undefined &&
nativeThreadId !== null &&
// A live process holds the conversation even before saving it.
(yield* Ref.get(queryContext))?.nativeThreadId !== nativeThreadId &&
// An unreadable transcript leaves the resume to the CLI.
(yield* hasEmptyClaudeTranscript(nativeThreadId).pipe(
Effect.orElseSucceed(() => false),
))
(yield* Ref.get(queryContext))?.nativeThreadId !== nativeThreadId
) {
return yield* new ProviderAdapter.ProviderAdapterProtocolError({
driver: CLAUDE_PROVIDER,
detail: `Claude session ${nativeThreadId} has no conversation to resume.`,
});
// An unreadable transcript leaves the resume to the CLI.
const transcript = yield* claudeTranscriptState(nativeThreadId).pipe(
Effect.orElseSucceed(() => "missing" as const),
);
if (transcript === "empty") {
return yield* new ProviderAdapter.ProviderAdapterProtocolError({
driver: CLAUDE_PROVIDER,
detail: `Claude session ${nativeThreadId} has no conversation to resume.`,
});
}
// A session imported from the terminal has history but no T3
// provider turns; opening it with sessionId fails "already in use".
if (transcript === "conversation") {
yield* Ref.update(openedNativeThreads, (current) =>
current.has(nativeThreadId) ? current : new Set(current).add(nativeThreadId),
);
}
}
const updatedAt = yield* DateTime.now;
return {
Expand Down
43 changes: 43 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@ import {
TurnItemId,
NodeId,
ProjectId,
ProviderDriverKind,
ProviderInstanceId,
} from "@t3tools/contracts";
import {
Expand Down Expand Up @@ -405,6 +406,12 @@ export interface OlderSettledThreadsListing {
readonly counts: ReadonlyArray<OrchestrationV2OlderSettledCount>;
}

export interface ProjectionNativeThreadOwner {
readonly threadId: ThreadId;
readonly providerInstanceId: ProviderInstanceId;
readonly deleted: boolean;
}

export interface ProjectionStoreV2Shape {
readonly searchThreadStream: (
input: OrchestrationV2SearchThreadInput,
Expand Down Expand Up @@ -475,6 +482,11 @@ export interface ProjectionStoreV2Shape {
readonly getThread: (
threadId: ThreadId,
) => Effect.Effect<OrchestrationV2AppThread, ProjectionStoreV2Error>;
/** App threads whose provider thread holds one of these native session ids. */
readonly findNativeThreadOwners: (options: {
readonly driver: ProviderDriverKind;
readonly nativeIds: ReadonlyArray<string>;
}) => Effect.Effect<ReadonlyMap<string, ProjectionNativeThreadOwner>, ProjectionStoreV2Error>;
readonly getLimitRecoveryCandidates: (options: {
readonly now: DateTime.Utc;
readonly autoResume: boolean;
Expand Down Expand Up @@ -6626,6 +6638,36 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
return rows[0] === undefined ? null : ThreadId.make(rows[0].thread_id);
}, Effect.mapError(shellSnapshotReadError));

const findNativeThreadOwners: ProjectionStoreV2Shape["findNativeThreadOwners"] = Effect.fn(
"ProjectionStore.findNativeThreadOwners",
)(function* ({ driver, nativeIds }) {
const owners = new Map<string, ProjectionNativeThreadOwner>();
if (nativeIds.length === 0) return owners;
const rows = yield* sql<{
readonly thread_id: string;
readonly provider_instance_id: string;
readonly native_id: string;
readonly deleted_at: string | null;
}>`
SELECT p.thread_id, p.provider_instance_id, t.deleted_at,
json_extract(p.payload_json, '$.nativeThreadRef.nativeId') AS native_id
FROM orchestration_v2_projection_provider_threads p
JOIN orchestration_v2_projection_threads t ON t.thread_id = p.thread_id
WHERE p.driver = ${driver}
AND json_extract(p.payload_json, '$.nativeThreadRef.nativeId') IN ${sql.in(nativeIds)}
`;
for (const row of rows) {
if (!owners.has(row.native_id)) {
owners.set(row.native_id, {
threadId: ThreadId.make(row.thread_id),
providerInstanceId: ProviderInstanceId.make(row.provider_instance_id),
deleted: row.deleted_at !== null,
});
}
}
return owners;
}, Effect.mapError(shellSnapshotReadError));

const listOlderSettledThreads: ProjectionStoreV2Shape["listOlderSettledThreads"] = Effect.fn(
"ProjectionStore.listOlderSettledThreads",
)(function* ({ before, projectIds, limit }) {
Expand Down Expand Up @@ -6772,6 +6814,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
listOlderSettledThreads: readOnReader(listOlderSettledThreads),
getThreadShell: readOnReader(getThreadShell),
findGuestShareThreadId: readOnReader(findGuestShareThreadId),
findNativeThreadOwners: readOnReader(findNativeThreadOwners),
getThread: readOnReader(getThread),
getSettlementCandidates: readOnReader(getSettlementCandidates),
getThreadsWithPullRequests: readOnReader(getThreadsWithPullRequests),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,7 @@ it.effect(
readShellSnapshot: () => Effect.die("unused readShellSnapshot"),
getThreadShell: () => Effect.die("unused getThreadShell"),
findGuestShareThreadId: () => Effect.die("unused findGuestShareThreadId"),
findNativeThreadOwners: () => Effect.die("unused findNativeThreadOwners"),
getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),
getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"),
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/orchestration-v2/runtimeLayer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,7 @@ const layerAgentSessionImporterProvided = AgentSessionImporter.layer.pipe(
layerProjectService,
layerOrchestratorProvided,
layerEventSink,
ProjectionStore.layer,
IdAllocator.layer,
ProviderSessionRuntime.layer,
),
Expand Down
Loading
Loading