Skip to content

Commit f423750

Browse files
lookmanrayslookmanrays
committed
Route project resume through Gateway Relay and connector
1 parent 4d4654b commit f423750

9 files changed

Lines changed: 225 additions & 12 deletions

File tree

‎internal/connector/project_requests.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,10 @@ func (c *Client) handleProjectRequest(ctx context.Context, cfg *Config, request
5252
service := localexec.NewService()
5353
report, err := service.CancelRun(ctx, localexec.RunOptions{BaseOptions: projectBaseOptions(cfg, project), RunID: tail[1]})
5454
return projectExecutionResponse(response, report, err)
55+
case len(tail) == 3 && tail[0] == "runs" && tail[2] == "resume" && request.Method == http.MethodPost:
56+
service := localexec.NewService()
57+
report, err := service.ResumeRun(ctx, localexec.RunOptions{BaseOptions: projectBaseOptions(cfg, project), RunID: tail[1]})
58+
return projectExecutionResponse(response, report, err)
5559
case len(tail) == 1 && tail[0] == "submit" && request.Method == http.MethodPost:
5660
opts, err := decodeProjectSubmit(projectBaseOptions(cfg, project), request.Body)
5761
if err != nil {

‎internal/connector/session_test.go‎

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,7 @@ func TestClientHandleProjectRequestStartsRunThroughLocalexec(t *testing.T) {
183183

184184
var daemon *httptest.Server
185185
var cancelled atomic.Bool
186+
var resumed atomic.Bool
186187
daemon = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
187188
switch r.URL.Path {
188189
case "/api/v1/instance":
@@ -197,8 +198,22 @@ func TestClientHandleProjectRequestStartsRunThroughLocalexec(t *testing.T) {
197198
case "/api/v1/runs/run-1":
198199
switch r.Method {
199200
case http.MethodPatch:
200-
cancelled.Store(true)
201-
_ = json.NewEncoder(w).Encode(domain.Run{ID: "run-1", ProjectID: "proj", State: domain.RunStateCancelled})
201+
var req struct {
202+
Action string `json:"action"`
203+
}
204+
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
205+
t.Fatalf("decode daemon patch action: %v", err)
206+
}
207+
switch req.Action {
208+
case "abort":
209+
cancelled.Store(true)
210+
_ = json.NewEncoder(w).Encode(domain.Run{ID: "run-1", ProjectID: "proj", State: domain.RunStateCancelled})
211+
case "resume":
212+
resumed.Store(true)
213+
_ = json.NewEncoder(w).Encode(domain.Run{ID: "run-1", ProjectID: "proj", State: domain.RunStateRunning, RecoveryNotes: "Resume requested by operator."})
214+
default:
215+
t.Fatalf("unexpected daemon patch action %q", req.Action)
216+
}
202217
case http.MethodGet:
203218
_ = json.NewEncoder(w).Encode(domain.Run{ID: "run-1", ProjectID: "proj", State: domain.RunStateCancelled})
204219
default:
@@ -256,6 +271,21 @@ func TestClientHandleProjectRequestStartsRunThroughLocalexec(t *testing.T) {
256271
if !strings.Contains(string(response.Body), `"run-1"`) || !strings.Contains(string(response.Body), `"cancelled"`) {
257272
t.Fatalf("unexpected project cancel response: %s", string(response.Body))
258273
}
274+
275+
response = client.handleRequest(context.Background(), relayproto.CommandRequest{
276+
RequestID: "req-project-resume",
277+
Method: http.MethodPost,
278+
Path: "/codencer/v1/projects/proj/runs/run-1/resume",
279+
})
280+
if response.StatusCode != http.StatusOK {
281+
t.Fatalf("expected ok resume response, got %+v", response)
282+
}
283+
if !resumed.Load() {
284+
t.Fatal("expected project resume request to reach daemon run resume endpoint")
285+
}
286+
if !strings.Contains(string(response.Body), `"run_resumed"`) || !strings.Contains(string(response.Body), `"running"`) {
287+
t.Fatalf("unexpected project resume response: %s", string(response.Body))
288+
}
259289
}
260290

261291
func TestClientHandleRequestRejectsUnsafeStepPath(t *testing.T) {

‎internal/gateway/api.go‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -825,6 +825,10 @@ func (s *Server) handleProjectByID(w http.ResponseWriter, r *http.Request) {
825825
s.handleProjectRunCancel(w, r, projectID, parts[2])
826826
return
827827
}
828+
if len(parts) == 4 && parts[1] == "runs" && parts[3] == "resume" {
829+
s.handleProjectRunResume(w, r, projectID, parts[2])
830+
return
831+
}
828832
if len(parts) > 1 {
829833
writeAPIError(w, http.StatusNotFound, "route_not_found", "project route not found")
830834
return
@@ -999,6 +1003,69 @@ func (s *Server) handleProjectRunCancel(w http.ResponseWriter, r *http.Request,
9991003
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "project_id": projectID, "run_id": runID, "run_history_id": runRecord.ID, "result": payload})
10001004
}
10011005

1006+
func (s *Server) handleProjectRunResume(w http.ResponseWriter, r *http.Request, projectID, runID string) {
1007+
if r.Method != http.MethodPost {
1008+
writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
1009+
return
1010+
}
1011+
principal, ok := s.authenticateConsoleAPI(w, r, []string{"projects:read", "runs:write"})
1012+
if !ok {
1013+
return
1014+
}
1015+
args := map[string]any{"project_id": projectID, "run_id": runID}
1016+
var req map[string]any
1017+
if r.Body != nil {
1018+
if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err != io.EOF {
1019+
writeAPIError(w, http.StatusBadRequest, "malformed_request", err.Error())
1020+
return
1021+
}
1022+
}
1023+
for key, value := range req {
1024+
args[key] = value
1025+
}
1026+
for _, key := range []string{"relay_profile_id", "machine_id", "host_label"} {
1027+
if value := strings.TrimSpace(r.URL.Query().Get(key)); value != "" {
1028+
args[key] = value
1029+
}
1030+
}
1031+
match, apiErr := s.resolveProject(r.Context(), principal, projectID, args, true)
1032+
if apiErr != nil {
1033+
s.recordGatewayAuditWithMetadata(r.Context(), principal, "resume_project_run_requested", "Run resume route resolution failed for project "+projectID+": "+apiErr.Code, map[string]any{"project_id": projectID, "run_id": runID})
1034+
writeAPIError(w, apiErr.Status, apiErr.Code, apiErr.Message)
1035+
return
1036+
}
1037+
auditMetadata := projectLifecycleAuditMetadata(match, args)
1038+
s.recordGatewayAuditWithMetadata(r.Context(), principal, "resume_project_run_requested", "Requested resume_project_run for run "+runID+" in project "+projectID, auditMetadata)
1039+
path, body, apiErr := resumeProjectRunRoute(args)
1040+
if apiErr != nil {
1041+
writeAPIError(w, apiErr.Status, apiErr.Code, apiErr.Message)
1042+
return
1043+
}
1044+
path = appendSelector(path, args)
1045+
_, response, apiErr := s.callRelay(r.Context(), match.Profile, http.MethodPost, path, body)
1046+
payload, apiErr := responsePayload(match.Profile, response, apiErr)
1047+
if apiErr != nil {
1048+
s.recordGatewayAuditWithMetadata(r.Context(), principal, "run_failed", "Run resume failed for project "+projectID+": "+apiErr.Code, auditMetadata)
1049+
writeAPIError(w, apiErr.Status, apiErr.Code, apiErr.Message)
1050+
return
1051+
}
1052+
runRecord, auditMetadata := s.refreshRunRecordFromLifecyclePayload(r.Context(), principal, match, args, payload, reportStatusForPayload(payload, "available"))
1053+
eventType := terminalAuditType(payload)
1054+
s.recordGatewayAuditWithMetadata(r.Context(), principal, eventType, terminalAuditSummary(projectID, payload), auditMetadata)
1055+
if eventType == "blocker" {
1056+
blockedMetadata := map[string]any{}
1057+
for key, value := range auditMetadata {
1058+
blockedMetadata[key] = value
1059+
}
1060+
blockedMetadata["operation"] = "resume_project_run"
1061+
blockedMetadata["status"] = "blocked"
1062+
blockedMetadata["blocker_type"] = "unsupported_operation"
1063+
s.recordGatewayAuditWithMetadata(r.Context(), principal, "resume_project_run_blocked", "Blocked unsupported resume_project_run for run "+runID+" in project "+projectID, blockedMetadata)
1064+
s.recordHumanInterruptAudit(r.Context(), principal, projectID, payload, auditMetadata)
1065+
}
1066+
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "project_id": projectID, "run_id": runID, "run_history_id": runRecord.ID, "result": payload})
1067+
}
1068+
10021069
func (s *Server) handleAuditEvents(w http.ResponseWriter, r *http.Request) {
10031070
if apiErr := s.requireStore(); apiErr != nil {
10041071
writeAPIError(w, apiErr.Status, apiErr.Code, apiErr.Message)

‎internal/gateway/gateway_test.go‎

Lines changed: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -361,16 +361,16 @@ func TestGatewayMCPAsyncLifecycleTools(t *testing.T) {
361361
"run_history_id": startedRunHistoryID,
362362
})
363363
resumeBody := mustJSON(t, resume)
364-
if !strings.Contains(resumeBody, `"type":"unsupported_operation"`) || !strings.Contains(resumeBody, `"operation":"resume_project_run"`) {
365-
t.Fatalf("expected structured resume capability blocker, got %s", resumeBody)
364+
if !strings.Contains(resumeBody, `"run_resumed"`) || !strings.Contains(resumeBody, `"state":"running"`) {
365+
t.Fatalf("expected forwarded resume response, got %s", resumeBody)
366366
}
367367
assertNoGatewayMCPLeak(t, resumeBody)
368368
resumeEvents := mcpToolCall(t, server.URL, session, "codencer.get_gateway_run_events", map[string]any{
369369
"run_history_id": startedRunHistoryID,
370370
"limit": 20,
371371
})
372372
resumeEventsBody := mustJSON(t, resumeEvents)
373-
for _, want := range []string{`"resume_project_run_requested"`, `"resume_project_run_blocked"`, `"operation":"resume_project_run"`, `"blocker_type":"unsupported_operation"`} {
373+
for _, want := range []string{`"resume_project_run_requested"`, `"run_resumed"`} {
374374
if !strings.Contains(resumeEventsBody, want) {
375375
t.Fatalf("resume run events missing %s: %s", want, resumeEventsBody)
376376
}
@@ -1094,6 +1094,29 @@ func newFakeRelay(t *testing.T, opts fakeRelayOptions) *fakeRelay {
10941094
"report_path": "/tmp/codencer/run-plans/run-async-gateway-test.json",
10951095
})
10961096
})
1097+
mux.HandleFunc("/api/v2/projects/codencer/runs/run-async-gateway-test/resume", func(w http.ResponseWriter, r *http.Request) {
1098+
requireRelayAuth(t, r)
1099+
if r.Method != http.MethodPost {
1100+
w.WriteHeader(http.StatusMethodNotAllowed)
1101+
return
1102+
}
1103+
writeTestJSON(t, w, map[string]any{
1104+
"ok": true,
1105+
"run_id": "run-async-gateway-test",
1106+
"status": "running",
1107+
"run": map[string]any{
1108+
"id": "run-async-gateway-test",
1109+
"state": "running",
1110+
},
1111+
"events": []map[string]any{{
1112+
"type": "run_resumed",
1113+
"run_id": "run-async-gateway-test",
1114+
"state": "running",
1115+
}},
1116+
"repo_root": "/Users/example/codencer",
1117+
"report_path": "/tmp/codencer/run-plans/run-async-gateway-test.json",
1118+
})
1119+
})
10971120
mux.HandleFunc("/api/v2/projects/codencer/run-plan", func(w http.ResponseWriter, r *http.Request) {
10981121
requireRelayAuth(t, r)
10991122
relay.lastMachineID = r.URL.Query().Get("machine_id")

‎internal/gateway/run_history.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,8 @@ func applyPayloadToRunRecord(record *RunRecord, payload any, reportStatus string
128128
}
129129

130130
func reportStatusForPayload(payload any, terminalStatus string) string {
131-
if terminalAuditType(payload) == "run_started" {
131+
eventType := terminalAuditType(payload)
132+
if eventType == "run_started" || eventType == "run_resumed" {
132133
return "pending"
133134
}
134135
return terminalStatus

‎internal/gateway/tools.go‎

Lines changed: 70 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ func buildTools(server *Server) map[string]Tool {
157157
"codencer.get_gateway_run_events": server.gatewayRunEventsTool(),
158158
"codencer.respond_to_human_interrupt": server.humanInterruptResponseTool(),
159159
"codencer.cancel_project_run": server.projectForwardTool("codencer.cancel_project_run", "Cancel a shared project run through the selected Gateway relay.", []string{"project_id", "run_id"}, cancelProjectRunRoute),
160-
"codencer.resume_project_run": server.unsupportedProjectLifecycleTool("codencer.resume_project_run", "Return an explicit capability blocker for project-run resume when the selected route cannot resume safely.", "resume_project_run"),
160+
"codencer.resume_project_run": server.projectForwardTool("codencer.resume_project_run", "Resume a shared project run through the selected Gateway relay when the route supports it.", []string{"project_id", "run_id"}, resumeProjectRunRoute),
161161
"codencer.get_blocker": {
162162
Name: "codencer.get_blocker",
163163
Description: "Read the blocker from a Gateway-routed project run report.",
@@ -263,9 +263,10 @@ func (s *Server) projectForwardTool(name, description string, required []string,
263263
return ToolResult{}, apiErr
264264
}
265265
path = appendSelector(path, args)
266-
if name == "codencer.cancel_project_run" {
266+
if name == "codencer.cancel_project_run" || name == "codencer.resume_project_run" {
267267
auditMetadata = projectLifecycleAuditMetadata(match, args)
268-
s.recordGatewayAuditWithMetadata(ctx, principal, "cancel_project_run_requested", "Requested cancel_project_run for run "+requiredStringValue(args, "run_id")+" in project "+projectID, auditMetadata)
268+
operation := lifecycleOperationForTool(name)
269+
s.recordGatewayAuditWithMetadata(ctx, principal, operation+"_requested", "Requested "+operation+" for run "+requiredStringValue(args, "run_id")+" in project "+projectID, auditMetadata)
269270
}
270271
method := http.MethodGet
271272
if body != nil {
@@ -280,7 +281,7 @@ func (s *Server) projectForwardTool(name, description string, required []string,
280281
eventType := "run_failed"
281282
if name == "codencer.get_run_report" {
282283
eventType = "report_read"
283-
} else if name == "codencer.cancel_project_run" {
284+
} else if name == "codencer.cancel_project_run" || name == "codencer.resume_project_run" {
284285
auditMetadata = projectLifecycleAuditMetadata(match, args)
285286
} else {
286287
runRecord.Status = "failed"
@@ -295,10 +296,22 @@ func (s *Server) projectForwardTool(name, description string, required []string,
295296
_ = record
296297
s.recordTerminalRunAuditOnce(ctx, principal, projectID, payload, metadata)
297298
s.recordGatewayAuditWithMetadata(ctx, principal, "report_read", "Read run report "+requiredStringValue(args, "run_id")+" for project "+projectID, metadata)
298-
} else if name == "codencer.cancel_project_run" {
299+
} else if name == "codencer.cancel_project_run" || name == "codencer.resume_project_run" {
299300
record, metadata := s.refreshRunRecordFromLifecyclePayload(ctx, principal, match, args, payload, reportStatusForPayload(payload, "available"))
300301
_ = record
301-
s.recordGatewayAuditWithMetadata(ctx, principal, terminalAuditType(payload), terminalAuditSummary(projectID, payload), metadata)
302+
eventType := terminalAuditType(payload)
303+
s.recordGatewayAuditWithMetadata(ctx, principal, eventType, terminalAuditSummary(projectID, payload), metadata)
304+
if name == "codencer.resume_project_run" && eventType == "blocker" {
305+
blockedMetadata := map[string]any{}
306+
for key, value := range metadata {
307+
blockedMetadata[key] = value
308+
}
309+
blockedMetadata["operation"] = "resume_project_run"
310+
blockedMetadata["status"] = "blocked"
311+
blockedMetadata["blocker_type"] = "unsupported_operation"
312+
s.recordGatewayAuditWithMetadata(ctx, principal, "resume_project_run_blocked", "Blocked unsupported resume_project_run for run "+requiredStringValue(args, "run_id")+" in project "+projectID, blockedMetadata)
313+
s.recordHumanInterruptAudit(ctx, principal, projectID, payload, metadata)
314+
}
302315
} else if recordRun {
303316
runRecord, auditMetadata = s.finishRunRecord(ctx, runRecord, payload, reportStatusForPayload(payload, "available"))
304317
if obj, ok := payload.(map[string]any); ok && runRecord.ID != "" {
@@ -625,12 +638,17 @@ func terminalAuditType(payload any) string {
625638
return "blocker"
626639
}
627640
}
641+
if payloadHasEventType(obj, "run_resumed") {
642+
return "run_resumed"
643+
}
628644
status := strings.ToLower(firstNonEmpty(
629645
stringValueFromAny(obj["status"]),
630646
nestedString(obj, "task", "status"),
631647
nestedString(obj, "run", "state"),
632648
))
633649
switch status {
650+
case "resumed":
651+
return "run_resumed"
634652
case "submitted", "started", "starting", "queued", "pending", "running", "in_progress", "validating":
635653
return "run_started"
636654
case "cancel_requested", "cancelling", "canceling":
@@ -660,13 +678,32 @@ func terminalAuditSummary(projectID string, payload any) string {
660678
outcome = "cancel requested"
661679
case "run_cancelled":
662680
outcome = "cancelled"
681+
case "run_resumed":
682+
outcome = "resumed"
663683
}
664684
if runID == "" {
665685
return "Run " + outcome + " for project " + projectID
666686
}
667687
return "Run " + runID + " " + outcome + " for project " + projectID
668688
}
669689

690+
func payloadHasEventType(obj map[string]any, eventType string) bool {
691+
events, ok := obj["events"].([]any)
692+
if !ok {
693+
return false
694+
}
695+
for _, event := range events {
696+
eventObj, ok := event.(map[string]any)
697+
if !ok {
698+
continue
699+
}
700+
if stringValueFromAny(eventObj["type"]) == eventType {
701+
return true
702+
}
703+
}
704+
return false
705+
}
706+
670707
func (s *Server) recordHumanInterruptAudit(ctx context.Context, principal *authPrincipal, projectID string, payload any, baseMetadata map[string]any) {
671708
interrupt := humanInterruptFromPayload(payload)
672709
if interrupt == nil {
@@ -854,6 +891,33 @@ func cancelProjectRunRoute(args map[string]any) (string, []byte, *apiError) {
854891
return "/api/v2/projects/" + projectID + "/runs/" + runID + "/cancel", body, nil
855892
}
856893

894+
func resumeProjectRunRoute(args map[string]any) (string, []byte, *apiError) {
895+
projectID, apiErr := requiredString(args, "project_id")
896+
if apiErr != nil {
897+
return "", nil, apiErr
898+
}
899+
runID, apiErr := requiredString(args, "run_id")
900+
if apiErr != nil {
901+
return "", nil, apiErr
902+
}
903+
payload := map[string]any{}
904+
copyOptional(payload, args, "reason")
905+
body, apiErr := jsonBody(payload)
906+
if apiErr != nil {
907+
return "", nil, apiErr
908+
}
909+
return "/api/v2/projects/" + projectID + "/runs/" + runID + "/resume", body, nil
910+
}
911+
912+
func lifecycleOperationForTool(name string) string {
913+
switch name {
914+
case "codencer.resume_project_run":
915+
return "resume_project_run"
916+
default:
917+
return "cancel_project_run"
918+
}
919+
}
920+
857921
func submitProjectTaskRoute(wait bool) func(map[string]any) (string, []byte, *apiError) {
858922
return func(args map[string]any) (string, []byte, *apiError) {
859923
projectID, apiErr := requiredString(args, "project_id")

‎internal/relay/mcp_tools.go‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -403,6 +403,23 @@ func addProjectMCPTools(server *mcpServer, tools map[string]mcpTool) {
403403
}
404404
return fmt.Sprintf("/api/v2/projects/%s/runs/%s/cancel", projectID, runID), body, nil
405405
})
406+
tools["codencer.resume_project_run"] = projectRouteTool(server, "codencer.resume_project_run", "Resume a run for a shared project when the selected route supports it.", "runs:write", objectSchema([]string{"project_id", "run_id"}, map[string]any{
407+
"project_id": stringSchema("Shared project identifier."),
408+
"run_id": stringSchema("Run identifier."),
409+
"reason": stringSchema("Optional planner/operator reason."),
410+
}), func(projectID string, args map[string]any) (string, []byte, *apiError) {
411+
runID, apiErr := requiredString(args, "run_id")
412+
if apiErr != nil {
413+
return "", nil, apiErr
414+
}
415+
payload := map[string]any{}
416+
copyOptional(payload, args, "reason")
417+
body, err := json.Marshal(payload)
418+
if err != nil {
419+
return "", nil, &apiError{Status: http.StatusBadRequest, Code: "malformed_request", Message: err.Error()}
420+
}
421+
return fmt.Sprintf("/api/v2/projects/%s/runs/%s/resume", projectID, runID), body, nil
422+
})
406423
tools["codencer.submit_project_task"] = projectSubmitTool(server, "codencer.submit_project_task", false)
407424
tools["codencer.submit_project_task_and_wait"] = projectSubmitTool(server, "codencer.submit_project_task_and_wait", true)
408425
tools["codencer.run_project_manifest"] = projectRouteTool(server, "codencer.run_project_manifest", "Run a project manifest sequentially for a shared project.", "runs:write", objectSchema([]string{"project_id"}, map[string]any{

‎internal/relay/router.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,9 @@ func (s *Server) handleResolvedProjectScoped(w http.ResponseWriter, r *http.Requ
196196
case len(parts) == 4 && parts[1] == "runs" && parts[3] == "cancel" && r.Method == http.MethodPost:
197197
body, _ := io.ReadAll(r.Body)
198198
s.projectProxyAndWrite(w, r, record, http.MethodPost, fmt.Sprintf("/codencer/v1/projects/%s/runs/%s/cancel", projectID, parts[2]), "", body, "cancel_project_run", "runs:write")
199+
case len(parts) == 4 && parts[1] == "runs" && parts[3] == "resume" && r.Method == http.MethodPost:
200+
body, _ := io.ReadAll(r.Body)
201+
s.projectProxyAndWrite(w, r, record, http.MethodPost, fmt.Sprintf("/codencer/v1/projects/%s/runs/%s/resume", projectID, parts[2]), "", body, "resume_project_run", "runs:write")
199202
case len(parts) == 2 && parts[1] == "submit" && r.Method == http.MethodPost:
200203
body, _ := io.ReadAll(r.Body)
201204
s.projectProxyAndWrite(w, r, record, http.MethodPost, fmt.Sprintf("/codencer/v1/projects/%s/submit", projectID), "", body, "submit_project_task", "steps:write")

0 commit comments

Comments
 (0)