Skip to content

Commit d0a40fd

Browse files
jonathanlabjonathanlab
committed
feat: local to cloud handoff
1 parent f84880f commit d0a40fd

18 files changed

Lines changed: 767 additions & 292 deletions

File tree

‎apps/code/src/main/services/git/service.ts‎

Lines changed: 61 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -104,13 +104,12 @@ function toUnifiedDiffPatch(
104104
@injectable()
105105
export class GitService extends TypedEventEmitter<GitServiceEvents> {
106106
private lastFetchTime = new Map<string, number>();
107-
private llmGateway: LlmGatewayService;
108107

109108
constructor(
110-
@inject(MAIN_TOKENS.LlmGatewayService) llmGateway: LlmGatewayService,
109+
@inject(MAIN_TOKENS.LlmGatewayService)
110+
private readonly llmGateway: LlmGatewayService,
111111
) {
112112
super();
113-
this.llmGateway = llmGateway;
114113
}
115114

116115
private async getStateSnapshot(
@@ -1086,85 +1085,71 @@ export class GitService extends TypedEventEmitter<GitServiceEvents> {
10861085

10871086
const [owner, repoName] = parts;
10881087

1089-
try {
1090-
const repoResult = await execGh([
1091-
"api",
1092-
`repos/${owner}/${repoName}`,
1093-
"--jq",
1094-
".default_branch",
1095-
]);
1088+
const repoResult = await execGh([
1089+
"api",
1090+
`repos/${owner}/${repoName}`,
1091+
"--jq",
1092+
".default_branch",
1093+
]);
10961094

1097-
if (repoResult.exitCode !== 0 || !repoResult.stdout.trim()) {
1098-
return [];
1099-
}
1100-
const defaultBranch = repoResult.stdout.trim();
1095+
if (repoResult.exitCode !== 0 || !repoResult.stdout.trim()) {
1096+
return [];
1097+
}
1098+
const defaultBranch = repoResult.stdout.trim();
11011099

1102-
const result = await execGh([
1103-
"api",
1104-
`repos/${owner}/${repoName}/compare/${defaultBranch}...${branch}`,
1105-
]);
1100+
const result = await execGh([
1101+
"api",
1102+
`repos/${owner}/${repoName}/compare/${defaultBranch}...${branch}`,
1103+
]);
11061104

1107-
if (result.exitCode !== 0) {
1108-
throw new Error(
1109-
`Failed to fetch branch files: ${result.stderr || result.error || "Unknown error"}`,
1110-
);
1105+
if (result.exitCode !== 0) {
1106+
throw new Error(
1107+
`Failed to fetch branch files: ${result.stderr || result.error || "Unknown error"}`,
1108+
);
1109+
}
1110+
1111+
const response = JSON.parse(result.stdout) as {
1112+
files?: Array<{
1113+
filename: string;
1114+
status: string;
1115+
previous_filename?: string;
1116+
additions: number;
1117+
deletions: number;
1118+
patch?: string;
1119+
}>;
1120+
};
1121+
const files = response.files;
1122+
1123+
if (!files) return [];
1124+
1125+
return files.map((f) => {
1126+
let status: ChangedFile["status"];
1127+
switch (f.status) {
1128+
case "added":
1129+
status = "added";
1130+
break;
1131+
case "removed":
1132+
status = "deleted";
1133+
break;
1134+
case "renamed":
1135+
status = "renamed";
1136+
break;
1137+
default:
1138+
status = "modified";
1139+
break;
11111140
}
11121141

1113-
const response = JSON.parse(result.stdout) as {
1114-
files?: Array<{
1115-
filename: string;
1116-
status: string;
1117-
previous_filename?: string;
1118-
additions: number;
1119-
deletions: number;
1120-
patch?: string;
1121-
}>;
1142+
return {
1143+
path: f.filename,
1144+
status,
1145+
originalPath: f.previous_filename,
1146+
linesAdded: f.additions,
1147+
linesRemoved: f.deletions,
1148+
patch: f.patch
1149+
? toUnifiedDiffPatch(f.patch, f.filename, f.previous_filename, status)
1150+
: undefined,
11221151
};
1123-
const files = response.files;
1124-
1125-
if (!files) return [];
1126-
1127-
return files.map((f) => {
1128-
let status: ChangedFile["status"];
1129-
switch (f.status) {
1130-
case "added":
1131-
status = "added";
1132-
break;
1133-
case "removed":
1134-
status = "deleted";
1135-
break;
1136-
case "renamed":
1137-
status = "renamed";
1138-
break;
1139-
default:
1140-
status = "modified";
1141-
break;
1142-
}
1143-
1144-
return {
1145-
path: f.filename,
1146-
status,
1147-
originalPath: f.previous_filename,
1148-
linesAdded: f.additions,
1149-
linesRemoved: f.deletions,
1150-
patch: f.patch
1151-
? toUnifiedDiffPatch(
1152-
f.patch,
1153-
f.filename,
1154-
f.previous_filename,
1155-
status,
1156-
)
1157-
: undefined,
1158-
};
1159-
});
1160-
} catch (error) {
1161-
log.warn("Failed to fetch branch changed files", {
1162-
repo,
1163-
branch,
1164-
error,
1165-
});
1166-
throw error;
1167-
}
1152+
});
11681153
}
11691154

11701155
public async generateCommitMessage(

‎apps/code/src/main/services/handoff/handoff-saga.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ function createDeps(overrides: Partial<HandoffSagaDeps> = {}): HandoffSagaDeps {
7373
getTaskRun: vi.fn().mockResolvedValue({
7474
log_url: "https://logs.example.com/run-1.ndjson",
7575
}),
76+
updateTaskRun: vi.fn().mockResolvedValue({}),
7677
}),
7778
applyTreeSnapshot: vi.fn().mockResolvedValue(undefined),
7879
applyGitCheckpoint: vi.fn().mockResolvedValue(undefined),
@@ -97,7 +98,6 @@ function createResumeState(
9798
conversation: [],
9899
latestSnapshot: null,
99100
latestGitCheckpoint: null,
100-
snapshotApplied: false,
101101
interrupted: false,
102102
logEntryCount: 0,
103103
...overrides,

‎apps/code/src/main/services/handoff/handoff-saga.ts‎

Lines changed: 10 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -6,29 +6,18 @@ import {
66
} from "@posthog/agent/resume";
77
import type * as AgentTypes from "@posthog/agent/types";
88
import { Saga, type SagaLogger } from "@posthog/shared";
9-
import type { WorkspaceMode } from "../../db/repositories/workspace-repository";
109
import type { SessionResponse } from "../agent/schemas";
11-
import type { HandoffStep } from "./schemas";
12-
13-
export interface HandoffSagaInput {
14-
taskId: string;
15-
runId: string;
16-
repoPath: string;
17-
apiHost: string;
18-
teamId: number;
19-
sessionId?: string;
20-
adapter?: "claude" | "codex";
21-
localGitState?: AgentTypes.HandoffLocalGitState;
22-
}
10+
import type { HandoffBaseDeps, HandoffExecuteInput } from "./schemas";
11+
12+
export type HandoffSagaInput = HandoffExecuteInput;
2313

2414
export interface HandoffSagaOutput {
2515
sessionId: string;
2616
snapshotApplied: boolean;
2717
conversationTurns: number;
2818
}
2919

30-
export interface HandoffSagaDeps {
31-
createApiClient(apiHost: string, teamId: number): PostHogAPIClient;
20+
export interface HandoffSagaDeps extends HandoffBaseDeps {
3221
applyTreeSnapshot(
3322
snapshot: AgentTypes.TreeSnapshotEvent,
3423
repoPath: string,
@@ -44,7 +33,6 @@ export interface HandoffSagaDeps {
4433
apiClient: PostHogAPIClient,
4534
localGitState?: AgentTypes.HandoffLocalGitState,
4635
): Promise<void>;
47-
updateWorkspaceMode(taskId: string, mode: WorkspaceMode): void;
4836
reconnectSession(params: {
4937
taskId: string;
5038
taskRunId: string;
@@ -63,9 +51,7 @@ export interface HandoffSagaDeps {
6351
localGitState?: AgentTypes.HandoffLocalGitState,
6452
): Promise<void>;
6553
seedLocalLogs(runId: string, logUrl: string): Promise<void>;
66-
killSession(taskRunId: string): Promise<void>;
6754
setPendingContext(taskRunId: string, context: string): void;
68-
onProgress(step: HandoffStep, message: string): void;
6955
}
7056

7157
export class HandoffSaga extends Saga<HandoffSagaInput, HandoffSagaOutput> {
@@ -97,6 +83,12 @@ export class HandoffSaga extends Saga<HandoffSagaInput, HandoffSagaOutput> {
9783

9884
const apiClient = this.deps.createApiClient(apiHost, teamId);
9985

86+
await this.readOnlyStep("update_run_environment", async () => {
87+
await apiClient.updateTaskRun(taskId, runId, {
88+
environment: "local",
89+
});
90+
});
91+
10092
const { resumeState, cloudLogUrl } = await this.readOnlyStep(
10193
"fetch_and_rebuild",
10294
async () => {

‎apps/code/src/main/services/handoff/handoff-to-cloud-saga.ts‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
import type * as AgentTypes from "@posthog/agent/types";
2+
import { Saga, type SagaLogger } from "@posthog/shared";
3+
import type { HandoffBaseDeps, HandoffToCloudExecuteInput } from "./schemas";
4+
5+
export type HandoffToCloudSagaInput = HandoffToCloudExecuteInput;
6+
7+
export interface HandoffToCloudSagaOutput {
8+
checkpointCaptured: boolean;
9+
snapshotCaptured: boolean;
10+
flushedLogEntryCount: number;
11+
}
12+
13+
export interface HandoffToCloudSagaDeps extends HandoffBaseDeps {
14+
captureGitCheckpoint(
15+
localGitState?: AgentTypes.HandoffLocalGitState,
16+
): Promise<AgentTypes.GitCheckpointEvent | null>;
17+
captureTreeSnapshot(): Promise<AgentTypes.TreeSnapshotEvent | null>;
18+
persistCheckpointToLog(
19+
checkpoint: AgentTypes.GitCheckpointEvent,
20+
): Promise<void>;
21+
persistSnapshotToLog(snapshot: AgentTypes.TreeSnapshotEvent): Promise<void>;
22+
flushLocalLogs(): Promise<number>;
23+
resumeRunInCloud(): Promise<void>;
24+
}
25+
26+
export class HandoffToCloudSaga extends Saga<
27+
HandoffToCloudSagaInput,
28+
HandoffToCloudSagaOutput
29+
> {
30+
readonly sagaName = "HandoffToCloudSaga";
31+
private deps: HandoffToCloudSagaDeps;
32+
33+
constructor(deps: HandoffToCloudSagaDeps, logger?: SagaLogger) {
34+
super(logger);
35+
this.deps = deps;
36+
}
37+
38+
protected async execute(
39+
input: HandoffToCloudSagaInput,
40+
): Promise<HandoffToCloudSagaOutput> {
41+
const { taskId, runId } = input;
42+
43+
let checkpointCaptured = false;
44+
let snapshotCaptured = false;
45+
46+
this.deps.onProgress(
47+
"capturing_checkpoint",
48+
"Capturing local git state...",
49+
);
50+
51+
const checkpoint = await this.readOnlyStep("capture_git_checkpoint", () =>
52+
this.deps.captureGitCheckpoint(input.localGitState),
53+
);
54+
55+
let persistedNotificationCount = 0;
56+
57+
if (checkpoint) {
58+
await this.readOnlyStep("persist_checkpoint_to_log", () =>
59+
this.deps.persistCheckpointToLog(checkpoint),
60+
);
61+
checkpointCaptured = true;
62+
persistedNotificationCount++;
63+
}
64+
65+
this.deps.onProgress("capturing_snapshot", "Capturing local file state...");
66+
67+
const snapshot = await this.readOnlyStep("capture_tree_snapshot", () =>
68+
this.deps.captureTreeSnapshot(),
69+
);
70+
71+
if (snapshot) {
72+
await this.readOnlyStep("persist_snapshot_to_log", () =>
73+
this.deps.persistSnapshotToLog(snapshot),
74+
);
75+
snapshotCaptured = true;
76+
persistedNotificationCount++;
77+
}
78+
79+
const localLogLineCount = await this.readOnlyStep("flush_local_logs", () =>
80+
this.deps.flushLocalLogs(),
81+
);
82+
const flushedLogEntryCount = localLogLineCount + persistedNotificationCount;
83+
84+
this.deps.onProgress("starting_cloud_run", "Starting cloud sandbox...");
85+
86+
await this.step({
87+
name: "start_cloud_run",
88+
execute: () => this.deps.resumeRunInCloud(),
89+
rollback: async () => {},
90+
});
91+
92+
this.deps.onProgress("stopping_agent", "Stopping local agent...");
93+
94+
await this.readOnlyStep("stop_local_agent", () =>
95+
this.deps.killSession(runId),
96+
);
97+
98+
await this.step({
99+
name: "update_workspace",
100+
execute: async () => {
101+
this.deps.updateWorkspaceMode(taskId, "cloud");
102+
},
103+
rollback: async () => {
104+
this.deps.updateWorkspaceMode(taskId, "local");
105+
},
106+
});
107+
108+
this.deps.onProgress("complete", "Handoff to cloud complete");
109+
110+
return { checkpointCaptured, snapshotCaptured, flushedLogEntryCount };
111+
}
112+
}

0 commit comments

Comments
 (0)