diff --git a/docs/convergence/protocol-registry.md b/docs/convergence/protocol-registry.md index 9cea1249f..80434e9f5 100644 --- a/docs/convergence/protocol-registry.md +++ b/docs/convergence/protocol-registry.md @@ -26,7 +26,7 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi | [exported-evidence-and-handoff](#exported-evidence-and-handoff) | `external-durable` | Evidence, verification, report, and handoff maintainers | 35 | true | | [extension-package-contracts](#extension-package-contracts) | `external-durable` | Skill, Plugin, Hook, and extension maintainers | 48 | true | | [github-pr-followup-observation](#github-pr-followup-observation) | `projection` | GitHub review and task follow-up maintainers | 1 | false | -| [http-openapi-contract](#http-openapi-contract) | `external-durable` | HTTP/OpenAPI and generated-client maintainers | 138 | true | +| [http-openapi-contract](#http-openapi-contract) | `external-durable` | HTTP/OpenAPI and generated-client maintainers | 139 | true | | [in-memory-token-session](#in-memory-token-session) | `ephemeral` | Credential and bootstrap maintainers | 1 | false | | [lsp-process-session](#lsp-process-session) | `ephemeral` | Code intelligence maintainers | 1 | false | | [mcp-interchange](#mcp-interchange) | `external-durable` | MCP client/server maintainers | 9 | true | @@ -986,7 +986,7 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi - `pr-followup-http-request-reader` (`v1`, active) at `internal/application/pr_followup.go` - `pr-followup-http-response-reader` (`v1`, active) at `web/src/api/pr-followup.ts` -
138 active identifiers +
139 active identifiers - `agent-code-tools.v1` - `agent_browser_close.v1` @@ -1102,6 +1102,7 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi - `thread_application_services.v1` - `thread_branch.v1` - `thread_change_summary.v1` +- `thread_change_undo.v1` - `thread_command_service_policy.v1` - `thread_context_compaction.v1` - `thread_context_compaction_operations.v1` diff --git a/docs/openapi.json b/docs/openapi.json index 9547a7aa6..5b3fa8b8d 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -28390,6 +28390,117 @@ "x-cyberagent-read-only": true } }, + "/api/v1/threads/{thread_id}/changes/undo": { + "post": { + "operationId": "undoThreadChanges", + "summary": "Preview or undo recorded changes for one Thread reply", + "description": "With confirm=false, checks current authority and conflicts without writing. With confirm=true, restores only receipt-proven paths to their state before this reply interval. One durable operation is bound to the exact Thread and source event; completed and partial results replay without applying again. Requires an idle physical working directory and current control authorization.", + "tags": [ + "Control" + ], + "parameters": [ + { + "name": "thread_id", + "in": "path", + "description": "Thread identity", + "required": true, + "schema": { + "maxLength": 256, + "minLength": 1, + "pattern": "^[^/\\\\\\x00-\\x1f\\x7f]+$", + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Control request accepted or idempotently replayed", + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "$ref": "#/components/schemas/ThreadChangeUndoResult" + }, + "request_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "const": "api.v1", + "type": "string" + } + }, + "required": [ + "version", + "request_id", + "data" + ], + "type": "object" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "403": { + "$ref": "#/components/responses/Forbidden" + }, + "404": { + "$ref": "#/components/responses/NotFound" + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "412": { + "$ref": "#/components/responses/FailedPrecondition" + }, + "413": { + "$ref": "#/components/responses/RequestEntityTooLarge" + }, + "414": { + "$ref": "#/components/responses/RequestTooLarge" + }, + "415": { + "$ref": "#/components/responses/UnsupportedMediaType" + }, + "429": { + "$ref": "#/components/responses/ResourceExhausted" + }, + "500": { + "$ref": "#/components/responses/InternalError" + }, + "503": { + "$ref": "#/components/responses/Unavailable" + }, + "504": { + "$ref": "#/components/responses/GatewayTimeout" + } + }, + "security": [ + { + "ControlBearerAuth": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ThreadChangeUndoRequest" + } + } + } + }, + "x-cyberagent-read-only": false + } + }, "/api/v1/threads/{thread_id}/command-service-policy": { "get": { "operationId": "getThreadCommandServicePolicy", @@ -68372,6 +68483,9 @@ "minLength": 1, "type": "string" }, + "undo": { + "$ref": "#/components/schemas/ThreadChangeUndoResult" + }, "version": { "type": "string" } @@ -68392,6 +68506,102 @@ ], "type": "object" }, + "ThreadChangeUndoConflict": { + "additionalProperties": false, + "properties": { + "path": { + "type": "string" + }, + "reason": { + "type": "string" + } + }, + "required": [ + "reason" + ], + "type": "object" + }, + "ThreadChangeUndoRequest": { + "additionalProperties": false, + "properties": { + "confirm": { + "type": "boolean" + }, + "source_event_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "type": "string" + } + }, + "required": [ + "version", + "source_event_id", + "confirm" + ], + "type": "object" + }, + "ThreadChangeUndoResult": { + "additionalProperties": false, + "properties": { + "conflicts": { + "items": { + "$ref": "#/components/schemas/ThreadChangeUndoConflict" + }, + "type": "array" + }, + "remaining_paths": { + "items": { + "type": "string" + }, + "type": "array" + }, + "replayed": { + "type": "boolean" + }, + "restored_paths": { + "items": { + "type": "string" + }, + "type": "array" + }, + "run_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "source_event_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "state": { + "type": "string" + }, + "thread_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "type": "string" + } + }, + "required": [ + "version", + "thread_id", + "source_event_id", + "run_id", + "state", + "conflicts", + "restored_paths", + "remaining_paths", + "replayed" + ], + "type": "object" + }, "ThreadCommandServicePolicy": { "additionalProperties": false, "properties": { diff --git a/docs/usage.md b/docs/usage.md index f5ff00b2c..31940062d 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -121,6 +121,11 @@ AI 回复下方的“本轮改动”卡片汇总本轮已应用的文件编辑 统计范围为文件编辑工具留下的应用记录；命令直接写入的文件不在此范围。敏感内容或不连续的 记录会标明差异与行数不可用，没有净改动时不显示卡片。 +卡片上的“撤销”会先检查当前文件，确认后将本轮记录的改动恢复到本轮开始时的内容。 +同一文件的多次编辑会一起撤销；其他路径与 Git 暂存区保持不变。有后续文件修改、任务仍在 +执行或编辑记录不完整时会显示原因。成功后显示“已撤销”，刷新页面后仍保留此状态； +请求中断时可“重新检查”结果。若恢复中途失败，会分别列出已恢复和未恢复的路径。 + Agent 建议固定任务要求时，问题会显示在输入框上方的独立卡片中。可选择“本次任务始终保留” 或“仅当前讨论”，也可直接撰写回复，再点击“发送”。自定义回复只有勾选“将这段回复固定到本次任务” 才会成为固定要求；“跳过”不固定内容，也不发送答复。回答失败时可重试原答复。 diff --git a/internal/application/run_worktree_service.go b/internal/application/run_worktree_service.go index d109c6907..9946e9824 100644 --- a/internal/application/run_worktree_service.go +++ b/internal/application/run_worktree_service.go @@ -136,6 +136,16 @@ func (s *RunWorktreeService) ReconcileWorkspaceCheckpoints(ctx context.Context) return nil, apperror.New(apperror.CodeConflict, "checkpoint recovery transaction workspace does not match its owner") } + if workspacecheckpoint.IsThreadUndo(transaction) { + scoped := *s.checkpoints + scoped.withRunWorkspaceResolver(func(ctx context.Context, runID string) (session.WorkspaceInfo, bool, error) { + if runID != run.ID { + return session.WorkspaceInfo{}, false, apperror.New(apperror.CodeConflict, "reply undo recovery Run changed") + } + return resolved.Workspace, true, nil + }) + return &scoped, nil + } // Older Drydock boundaries used the ordinary cursor table. Preserve that // recorded identity when it still points at this transaction; never adopt // a source cursor or a different Drydock boundary into the new file cursor. diff --git a/internal/application/thread_change_undo.go b/internal/application/thread_change_undo.go new file mode 100644 index 000000000..168bb7a83 --- /dev/null +++ b/internal/application/thread_change_undo.go @@ -0,0 +1,556 @@ +package application + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "sort" + "time" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/fileedit" + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/redact" + "cyberagent-workbench/internal/runmutation" + "cyberagent-workbench/internal/workspacecheckpoint" +) + +const ThreadChangeUndoVersion = "thread_change_undo.v1" + +type ThreadChangeUndoRequest struct { + Version string `json:"version"` + SourceEventID string `json:"source_event_id"` + Confirm bool `json:"confirm"` +} + +type ThreadChangeUndoConflict struct { + Path string `json:"path,omitempty"` + Reason string `json:"reason"` +} + +type ThreadChangeUndoResult struct { + Version string `json:"version"` + ThreadID string `json:"thread_id"` + SourceEventID string `json:"source_event_id"` + RunID string `json:"run_id"` + State string `json:"state"` + Conflicts []ThreadChangeUndoConflict `json:"conflicts"` + RestoredPaths []string `json:"restored_paths"` + RemainingPaths []string `json:"remaining_paths"` + Replayed bool `json:"replayed"` +} + +type ThreadChangeUndoStore interface { + ThreadChangeStore + WorkspaceCheckpointStore + GetThread(context.Context, string) (domain.Thread, error) + CheckThreadChangeUndoIdle(context.Context, string, string, string) error + ClaimThreadChangeUndo(context.Context, workspacecheckpoint.Transaction) (workspacecheckpoint.Transaction, bool, error) +} + +type ThreadChangeUndoService struct { + store ThreadChangeUndoStore + capabilities domain.ExecutionPermissionRuntimeCapabilities + drydocks *RunWorktreeService +} + +func NewThreadChangeUndoService(store ThreadChangeUndoStore, capabilities domain.ExecutionPermissionRuntimeCapabilities) *ThreadChangeUndoService { + return &ThreadChangeUndoService{store: store, capabilities: capabilities} +} + +func (s *ThreadChangeUndoService) WithDrydocks(drydocks *RunWorktreeService) *ThreadChangeUndoService { + s.drydocks = drydocks + return s +} + +func threadUndoDigest(threadID, eventID string) string { + return runmutation.Fingerprint(ThreadChangeUndoVersion, threadID, eventID) +} + +// Status only reads the durable receipt; it deliberately does not inspect live +// files. Preview performs the fresh filesystem and authorization checks. +func (s *ThreadChangeUndoService) Status(ctx context.Context, summary ThreadChangeSummary) (ThreadChangeUndoResult, error) { + r := ThreadChangeUndoResult{Version: ThreadChangeUndoVersion, ThreadID: summary.ThreadID, + SourceEventID: summary.SourceEventID, RunID: summary.RunID, State: "available", + Conflicts: []ThreadChangeUndoConflict{}, RestoredPaths: []string{}, RemainingPaths: []string{}} + if summary.FileCount == 0 { + r.State = "unavailable" + } + t, found, err := s.store.GetWorkspaceCheckpointTransactionByOperation(ctx, threadUndoDigest(summary.ThreadID, summary.SourceEventID)) + if err != nil { + return r, apperror.Normalize(err) + } + if !found { + return r, nil + } + if !workspacecheckpoint.IsThreadUndo(t) || t.TriggerReceiptID != workspacecheckpoint.ThreadUndoReceiptPrefix+summary.SourceEventID { + return r, threadChangeBindingError() + } + r.Replayed = true + switch t.Status { + case workspacecheckpoint.TransactionCompleted: + r.State = "completed" + case workspacecheckpoint.TransactionPrepared, workspacecheckpoint.TransactionApplying: + r.State = "applying" + default: + r.State = "partial" + } + if t.AfterCheckpointID != "" { + after, e := s.store.GetWorkspaceCheckpointSnapshot(ctx, t.AfterCheckpointID) + if e != nil { + return r, apperror.Normalize(e) + } + target, e := s.store.GetWorkspaceCheckpointSnapshot(ctx, t.TargetCheckpointID) + if e != nil { + return r, apperror.Normalize(e) + } + classifyThreadUndoPaths(&r, target, after) + } else { + target, e := s.store.GetWorkspaceCheckpointSnapshot(ctx, t.TargetCheckpointID) + if e != nil { + return r, apperror.Normalize(e) + } + for _, entry := range target.Entries { + r.RemainingPaths = append(r.RemainingPaths, safeThreadActivityPath(entry.Path)) + } + } + if r.State == "partial" { + r.Conflicts = append(r.Conflicts, ThreadChangeUndoConflict{Reason: "restore_failed"}) + } + return r, nil +} + +type threadUndoPath struct { + beforeHash, beforeText, afterHash, afterText, modePath, sourceEditID string + beforeMode uint32 +} + +// Undo restores only paths proven by exact applied receipts before this reply. +// The existing checkpoint writer supplies path confinement and per-path CAS; +// its immutable transaction supplies durable replay and execution exclusion. +func (s *ThreadChangeUndoService) Undo(ctx context.Context, threadID string, request ThreadChangeUndoRequest) (ThreadChangeUndoResult, error) { + if request.Version != ThreadChangeUndoVersion { + return ThreadChangeUndoResult{}, apperror.New(apperror.CodeInvalidArgument, "unsupported Thread change undo version") + } + summary, err := NewThreadChangeService(s.store).Summary(ctx, threadID, request.SourceEventID) + if err != nil { + return ThreadChangeUndoResult{}, err + } + r, err := s.Status(ctx, summary) + if err != nil || r.State == "completed" || r.State == "unavailable" { + return r, err + } + digest := threadUndoDigest(threadID, request.SourceEventID) + t, found, err := s.store.GetWorkspaceCheckpointTransactionByOperation(ctx, digest) + if err != nil { + return r, apperror.Normalize(err) + } + if found { + return r, nil + } + thread, err := s.store.GetThread(ctx, threadID) + if err != nil { + return r, apperror.Normalize(err) + } + run, err := s.store.GetRun(ctx, thread.LastRunID) + if err != nil { + return r, apperror.Normalize(err) + } + mission, err := s.store.GetMission(ctx, run.MissionID) + if err != nil { + return r, apperror.Normalize(err) + } + files, err := ResolveRunFileWorkspace(ctx, s.store, run, mission, s.drydocks) + if err != nil { + return threadUndoBlocked(r, "workspace_changed"), nil + } + if files.Drydock != nil { + if err := requireCurrentRunFileDrydock(ctx, s.store, run.ID, *files.Drydock); err != nil { + return threadUndoBlocked(r, "workspace_changed"), nil + } + } + if err := requireSharedCommandServicesStopped(ctx, s.store, files.Workspace.RootPath); err != nil { + return threadUndoBlocked(r, "run_busy"), nil + } + if thread.Status != domain.ThreadActive { + return threadUndoBlocked(r, "run_busy"), nil + } + if err := s.store.CheckThreadChangeUndoIdle(ctx, threadID, run.ID, t.ID); err != nil { + return threadUndoBlocked(r, "run_busy"), nil + } + mode, err := s.store.GetRunMode(ctx, run.ID) + if err != nil { + return r, apperror.Normalize(err) + } + permission, err := s.store.GetRunExecutionPermission(ctx, run.ID) + if err != nil { + return r, apperror.Normalize(err) + } + if mode.RunID != run.ID || mode.MissionID != mission.ID || mode.Surface != domain.ExecutionSurfaceCode || mode.Phase != domain.ExecutionPhaseDeliver || + permission.RunID != run.ID || permission.MissionID != mission.ID || !s.capabilities.AllowsSnapshot(permission) || + (permission.Mode.IsApprovalMode() && !s.capabilities.OperatorApprovalEnabled) { + return threadUndoBlocked(r, "permission_denied"), nil + } + lease, leased, err := s.store.GetRunExecutionLease(ctx, run.ID) + if err != nil { + return r, apperror.Normalize(err) + } + if leased && lease.ActiveAt(time.Now().UTC()) { + return threadUndoBlocked(r, "run_busy"), nil + } + paths, err := s.paths(ctx, summary, files.Workspace.ID, run) + if err != nil { + return threadUndoBlocked(r, "incomplete_history"), nil + } + observed, err := workspacecheckpoint.Capture(ctx, workspacecheckpoint.CaptureRequest{ID: idgen.New("thread-undo-observed"), RunID: run.ID, MissionID: mission.ID, SessionID: run.SessionID, + WorkspaceID: files.Workspace.ID, WorkspaceRoot: files.Workspace.RootPath, Trigger: workspacecheckpoint.TriggerRewindPreflight, Phase: workspacecheckpoint.PhasePreflight, + TriggerReceiptID: workspacecheckpoint.ThreadUndoReceiptPrefix + request.SourceEventID, CreatedAt: time.Now().UTC()}) + if err != nil { + return r, apperror.Normalize(err) + } + if err := validateThreadUndoObservation(files.Workspace.RootPath, observed, paths); err != nil { + return threadUndoBlocked(r, "incomplete_history"), nil + } + if err := s.resolveThreadUndoModes(ctx, summary.RunID, files.Workspace.ID, observed, paths); err != nil { + return threadUndoBlocked(r, "incomplete_history"), nil + } + observed = scopeThreadUndoSnapshot(observed, paths) + before := threadUndoSnapshot(observed, paths, false, idgen.New("thread-undo-before")) + target := threadUndoSnapshot(observed, paths, true, idgen.New("thread-undo-target")) + preview, err := workspacecheckpoint.PreviewRestore(before, target, observed) + if err != nil { + return r, apperror.Normalize(err) + } + classifyThreadUndoPaths(&r, target, observed) + r.Conflicts = []ThreadChangeUndoConflict{} + for _, c := range preview.Conflicts { + r.Conflicts = append(r.Conflicts, ThreadChangeUndoConflict{Path: safeThreadActivityPath(c.Path), Reason: "files_changed"}) + } + if len(r.Conflicts) > 0 { + r.State = "conflict" + return r, nil + } + r.State = "available" + if !request.Confirm { + return r, nil + } + for _, snapshot := range []workspacecheckpoint.Snapshot{before, target} { + if _, _, err = s.store.CreateWorkspaceCheckpoint(ctx, snapshot); err != nil { + return r, apperror.Normalize(err) + } + } + now := time.Now().UTC() + t = workspacecheckpoint.Transaction{ID: "thread-undo-" + digest[:40], ProtocolVersion: workspacecheckpoint.ProtocolVersion, OperationKeyDigest: digest, + RequestFingerprint: runmutation.Fingerprint(ThreadChangeUndoVersion, threadID, request.SourceEventID, run.ID, files.Workspace.ID), RunID: run.ID, WorkspaceID: files.Workspace.ID, + Kind: workspacecheckpoint.TransactionUndo, TriggerReceiptID: workspacecheckpoint.ThreadUndoReceiptPrefix + request.SourceEventID, + BeforeCheckpointID: before.Checkpoint.ID, TargetCheckpointID: target.Checkpoint.ID, Status: workspacecheckpoint.TransactionPrepared, + RecoveryLevel: workspacecheckpoint.RecoveryComplete, ConflictJSON: "[]", CreatedAt: now, UpdatedAt: now} + t, r.Replayed, err = s.store.CreateWorkspaceCheckpointTransaction(ctx, t) + if err != nil { + return r, apperror.Normalize(err) + } + if r.Replayed { + return s.Status(ctx, summary) + } + var claimed bool + t, claimed, err = s.store.ClaimThreadChangeUndo(ctx, t) + if err != nil { + return r, apperror.Normalize(err) + } + if !claimed { + return s.Status(ctx, summary) + } + _, applyErr := workspacecheckpoint.ApplyRestore(ctx, files.Workspace.RootPath, before, target, observed) + // Complete against a context that survives a disconnected HTTP caller. + finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer cancel() + after, captureErr := workspacecheckpoint.Capture(finishCtx, workspacecheckpoint.CaptureRequest{ID: idgen.New("thread-undo-after"), RunID: run.ID, MissionID: mission.ID, SessionID: run.SessionID, + WorkspaceID: files.Workspace.ID, WorkspaceRoot: files.Workspace.RootPath, Trigger: workspacecheckpoint.TriggerRewindResult, Phase: workspacecheckpoint.PhaseAfter, + TriggerReceiptID: t.ID, CreatedAt: time.Now().UTC()}) + if captureErr != nil { + r.State = "partial" + r.Conflicts = append(r.Conflicts, ThreadChangeUndoConflict{Reason: "restore_failed"}) + return r, errors.Join(applyErr, captureErr) + } + if err := validateThreadUndoObservation(files.Workspace.RootPath, after, paths); err != nil { + return r, apperror.Normalize(err) + } + after = scopeThreadUndoSnapshot(after, paths) + if _, _, err = s.store.CreateWorkspaceCheckpoint(finishCtx, after); err != nil { + return r, apperror.Normalize(err) + } + classifyThreadUndoPaths(&r, target, after) + verification, verifyErr := workspacecheckpoint.PreviewRestore(target, target, after) + if verifyErr != nil || len(verification.Conflicts) > 0 { + applyErr = errors.Join(applyErr, verifyErr, errors.New("reply undo final state changed")) + } + t.AfterCheckpointID = after.Checkpoint.ID + if applyErr == nil && len(r.RemainingPaths) == 0 { + r.State = "completed" + t.Status = workspacecheckpoint.TransactionCompleted + } else { + r.State = "partial" + r.Conflicts = append(r.Conflicts, ThreadChangeUndoConflict{Reason: "restore_failed"}) + t.Status = workspacecheckpoint.TransactionInterrupted + t.ErrorCode = "restore_failed" + } + t.UpdatedAt = time.Now().UTC() + t.CompletedAt = &t.UpdatedAt + if _, _, err = s.store.UpdateWorkspaceCheckpointTransaction(finishCtx, t); err != nil { + return r, apperror.Normalize(err) + } + return r, nil +} + +func threadUndoBlocked(r ThreadChangeUndoResult, reason string) ThreadChangeUndoResult { + r.State = "unavailable" + r.Conflicts = append(r.Conflicts, ThreadChangeUndoConflict{Reason: reason}) + return r +} + +func (s *ThreadChangeUndoService) paths(ctx context.Context, summary ThreadChangeSummary, workspaceID string, currentRun domain.Run) (map[string]threadUndoPath, error) { + source, facts, err := s.store.ListThreadReplyFileEditEvents(ctx, summary.ThreadID, summary.SourceEventID, MaxThreadChangeEdits+1) + if err != nil { + return nil, err + } + if len(facts) > MaxThreadChangeEdits { + return nil, threadChangeBindingError() + } + run, err := s.store.GetRun(ctx, source.RunID) + if err != nil { + return nil, err + } + paths := map[string]threadUndoPath{} + seen := map[string]bool{} + set := func(path, from, fromText, to, toText, sourceEditID string) error { + if !validProposalSourcePath(path) || !validPersistedOriginal(from, fromText) || !validPersistedOriginal(to, toText) { + return threadChangeBindingError() + } + if old, ok := paths[path]; ok { + if old.afterHash != from { + return threadChangeBindingError() + } + old.afterHash, old.afterText = to, toText + paths[path] = old + } else { + paths[path] = threadUndoPath{beforeHash: from, beforeText: fromText, afterHash: to, afterText: toText, modePath: path, sourceEditID: sourceEditID} + } + return nil + } + for _, fact := range facts { + if seen[fact.SubjectID] { + continue + } + seen[fact.SubjectID] = true + e, err := s.store.GetFileEdit(ctx, fact.SubjectID) + if err != nil { + return nil, err + } + if err = NewThreadChangeService(s.store).validateEdit(ctx, run, fact, e); err != nil { + return nil, err + } + if e.WorkspaceID != workspaceID || e.SecretsRedacted || redact.String(e.OriginalText) != e.OriginalText || redact.String(e.ProposedText) != e.ProposedText { + return nil, threadChangeBindingError() + } + if err = checkRunProjectPaths(currentRun, e.Path, e.DestinationPath); err != nil { + return nil, err + } + if err = checkRunProjectPaths(run, e.Path, e.DestinationPath); err != nil { + return nil, err + } + if err = set(e.Path, e.OriginalHash, e.OriginalText, e.ProposedHash, e.ProposedText, e.ID); err != nil { + return nil, err + } + if e.Operation == fileedit.OperationMove { + if e.ProposedHash != "missing" || e.DestinationOriginalHash != "missing" || e.DestinationProposedHash != e.OriginalHash { + return nil, threadChangeBindingError() + } + if err = set(e.DestinationPath, "missing", "", e.DestinationProposedHash, e.OriginalText, e.ID); err != nil { + return nil, err + } + for path, p := range paths { + if p.modePath == e.Path { + p.modePath = e.DestinationPath + paths[path] = p + } + } + } + } + return paths, nil +} + +func validateThreadUndoObservation(root string, snapshot workspacecheckpoint.Snapshot, paths map[string]threadUndoPath) error { + captured := map[string]string{} + for _, e := range snapshot.Entries { + captured[e.Path] = e.WorktreeSHA256 + } + for path := range paths { + hash, err := fileedit.CurrentCreateHash(root, path) + if err != nil { + return err + } + observed := captured[path] + if observed == "" { + observed = "missing" + } + if hash != observed { + return apperror.New(apperror.CodeConflict, "reply path was not captured exactly") + } + } + return nil +} + +func (s *ThreadChangeUndoService) resolveThreadUndoModes(ctx context.Context, runID, workspaceID string, observed workspacecheckpoint.Snapshot, paths map[string]threadUndoPath) error { + modes := map[string]uint32{} + for _, e := range observed.Entries { + modes[e.Path] = e.Mode + } + var checkpoints []workspacecheckpoint.Checkpoint + for path, p := range paths { + if p.beforeHash == "missing" { + continue + } + p.beforeMode = modes[p.modePath] + if p.beforeMode == 0 { + if checkpoints == nil { + var err error + checkpoints, err = s.store.ListWorkspaceCheckpoints(ctx, runID, workspaceCheckpointListLimit) + if err != nil { + return err + } + } + for _, c := range checkpoints { + if c.WorkspaceID != workspaceID || c.Trigger != workspacecheckpoint.TriggerFileTool || c.Phase != workspacecheckpoint.PhaseBefore || c.TriggerReceiptID != p.sourceEditID { + continue + } + snapshot, err := s.store.GetWorkspaceCheckpointSnapshot(ctx, c.ID) + if err != nil { + return err + } + for _, e := range snapshot.Entries { + if e.Path == path && e.WorktreeSHA256 == p.beforeHash && e.Kind == workspacecheckpoint.EntryFile { + p.beforeMode = e.Mode + break + } + } + if p.beforeMode != 0 { + break + } + } + if p.beforeMode == 0 { + return threadChangeBindingError() + } + } + paths[path] = p + } + return nil +} + +func scopeThreadUndoSnapshot(snapshot workspacecheckpoint.Snapshot, paths map[string]threadUndoPath) workspacecheckpoint.Snapshot { + entries := []workspacecheckpoint.Entry{} + for _, entry := range snapshot.Entries { + if _, ok := paths[entry.Path]; ok { + entries = append(entries, entry) + } + } + snapshot.Entries = entries + // Preserve the index digest for drift detection, without persisting or + // restoring index content as part of a reply's selected file paths. + snapshot.Checkpoint.IndexBlobSHA256 = "" + return sealThreadUndoSnapshot(snapshot) +} + +func threadUndoSnapshot(observed workspacecheckpoint.Snapshot, paths map[string]threadUndoPath, target bool, id string) workspacecheckpoint.Snapshot { + snapshot := observed + snapshot.Checkpoint.ID = id + snapshot.Checkpoint.Phase = workspacecheckpoint.PhaseStandalone + snapshot.Entries = []workspacecheckpoint.Entry{} + snapshot.Blobs = []workspacecheckpoint.Blob{} + modes := map[string]uint32{} + for _, e := range observed.Entries { + modes[e.Path] = e.Mode + } + for path, p := range paths { + hash, text := p.afterHash, p.afterText + if target { + hash, text = p.beforeHash, p.beforeText + } + mode := modes[path] + if target && p.beforeMode != 0 { + mode = p.beforeMode + } + if mode == 0 { + mode = 0600 + } + e := workspacecheckpoint.Entry{Path: path, Kind: workspacecheckpoint.EntryFile, Mode: mode, WorktreeSHA256: hash, Recoverable: true} + if hash == "missing" { + e.State = workspacecheckpoint.StateMissing + e.StoragePolicy = workspacecheckpoint.StorageMissing + e.Mode = 0 + } else { + e.State = workspacecheckpoint.StatePresent + e.StoragePolicy = workspacecheckpoint.StorageStored + e.Size = int64(len(text)) + e.BlobSHA256 = hash + snapshot.Blobs = append(snapshot.Blobs, workspacecheckpoint.Blob{SHA256: hash, Content: []byte(text), CreatedAt: snapshot.Checkpoint.CreatedAt}) + } + snapshot.Entries = append(snapshot.Entries, e) + } + // Undo never restores the index; preserve its digest without storing a blob. + snapshot.Checkpoint.IndexBlobSHA256 = "" + return sealThreadUndoSnapshot(snapshot) +} + +func sealThreadUndoSnapshot(snapshot workspacecheckpoint.Snapshot) workspacecheckpoint.Snapshot { + sort.Slice(snapshot.Entries, func(i, j int) bool { return snapshot.Entries[i].Path < snapshot.Entries[j].Path }) + needed := map[string]bool{} + for _, e := range snapshot.Entries { + if e.BlobSHA256 != "" { + needed[e.BlobSHA256] = true + } + } + if snapshot.Checkpoint.IndexBlobSHA256 != "" { + needed[snapshot.Checkpoint.IndexBlobSHA256] = true + } + blobs := []workspacecheckpoint.Blob{} + var stored int64 + for _, b := range snapshot.Blobs { + if needed[b.SHA256] { + blobs = append(blobs, b) + stored += int64(len(b.Content)) + delete(needed, b.SHA256) + } + } + snapshot.Blobs = blobs + data, _ := json.Marshal(snapshot.Entries) + hash := sha256.Sum256(data) + snapshot.Checkpoint.ManifestSHA256 = hex.EncodeToString(hash[:]) + snapshot.Checkpoint.EntryCount = len(snapshot.Entries) + snapshot.Checkpoint.StoredBytes = stored + snapshot.Checkpoint.Title = workspacecheckpoint.ThreadUndoSnapshotTitle + snapshot.Checkpoint.RecoveryLevel = workspacecheckpoint.RecoveryUnavailable + return snapshot +} + +func classifyThreadUndoPaths(r *ThreadChangeUndoResult, target, observed workspacecheckpoint.Snapshot) { + live := map[string]string{} + for _, e := range observed.Entries { + live[e.Path] = e.WorktreeSHA256 + } + r.RestoredPaths = []string{} + r.RemainingPaths = []string{} + for _, e := range target.Entries { + hash := live[e.Path] + if hash == "" { + hash = "missing" + } + if hash == e.WorktreeSHA256 { + r.RestoredPaths = append(r.RestoredPaths, safeThreadActivityPath(e.Path)) + } else { + r.RemainingPaths = append(r.RemainingPaths, safeThreadActivityPath(e.Path)) + } + } +} diff --git a/internal/application/thread_change_undo_recovery.go b/internal/application/thread_change_undo_recovery.go new file mode 100644 index 000000000..3cdd36891 --- /dev/null +++ b/internal/application/thread_change_undo_recovery.go @@ -0,0 +1,43 @@ +package application + +import ( + "context" + "time" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/workspacecheckpoint" +) + +// Restart recovery records the actual selected paths and releases the existing +// execution reservation. It never guesses that a partial undo succeeded. +func (s *WorkspaceCheckpointService) reconcileThreadChangeUndo(ctx context.Context, binding workspaceCheckpointBinding, t workspacecheckpoint.Transaction) error { + before, err := s.store.GetWorkspaceCheckpointSnapshot(ctx, t.BeforeCheckpointID) + if err != nil { + return apperror.Normalize(err) + } + paths := map[string]threadUndoPath{} + for _, e := range before.Entries { + paths[e.Path] = threadUndoPath{} + } + after, err := workspacecheckpoint.Capture(ctx, s.workspaceCaptureRequest(binding, workspacecheckpoint.CaptureRequest{ + ID: idgen.New("thread-undo-recovered"), Trigger: workspacecheckpoint.TriggerRewindResult, Phase: workspacecheckpoint.PhaseAfter, + TriggerReceiptID: t.ID, CreatedAt: time.Now().UTC()})) + if err != nil { + return apperror.Normalize(err) + } + if err := validateThreadUndoObservation(binding.workspace.RootPath, after, paths); err != nil { + return apperror.Normalize(err) + } + after = scopeThreadUndoSnapshot(after, paths) + if _, _, err = s.store.CreateWorkspaceCheckpoint(ctx, after); err != nil { + return apperror.Normalize(err) + } + t.Status = workspacecheckpoint.TransactionInterrupted + t.ErrorCode = "process_restart_reconciliation" + t.AfterCheckpointID = after.Checkpoint.ID + t.UpdatedAt = time.Now().UTC() + t.CompletedAt = &t.UpdatedAt + _, _, err = s.store.UpdateWorkspaceCheckpointTransaction(ctx, t) + return apperror.Normalize(err) +} diff --git a/internal/application/thread_change_undo_recovery_test.go b/internal/application/thread_change_undo_recovery_test.go new file mode 100644 index 000000000..f18aa5ea4 --- /dev/null +++ b/internal/application/thread_change_undo_recovery_test.go @@ -0,0 +1,104 @@ +package application + +import ( + "os" + "path/filepath" + "testing" + "time" + + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/fileedit" + "cyberagent-workbench/internal/runmutation" + "cyberagent-workbench/internal/session" + "cyberagent-workbench/internal/workspacecheckpoint" +) + +func TestThreadChangeUndoClaimAndDrydockRestartRecovery(t *testing.T) { + f := newDrydockApplicationFixture(t, "reply undo recovery") + checkpoints, err := NewWorkspaceCheckpointService(f.state, domain.ExecutionPermissionRuntimeCapabilities{OperatorApprovalEnabled: true}) + if err != nil { + t.Fatal(err) + } + f.service.WithCheckpointService(checkpoints) + w := mustCreateDrydock(t, f) + writeDrydockTestFile(t, filepath.Join(w.Path, "tracked.txt"), "changed\n") + if _, err := f.state.SaveSessionMessage(t.Context(), session.NewMessage(f.run.SessionID, "assistant", "done")); err != nil { + t.Fatal(err) + } + facts, err := f.state.ListRunEvents(t.Context(), f.run.ID) + if err != nil { + t.Fatal(err) + } + sourceID := "" + for _, fact := range facts { + if fact.Type == events.SessionMessageEvent { + sourceID = fact.EventID + } + } + observed, err := workspacecheckpoint.Capture(t.Context(), workspacecheckpoint.CaptureRequest{ID: "undo-observed", RunID: f.run.ID, MissionID: f.run.MissionID, SessionID: f.run.SessionID, + WorkspaceID: w.WorkspaceID, WorkspaceRoot: w.Path, Trigger: workspacecheckpoint.TriggerRewindPreflight, Phase: workspacecheckpoint.PhasePreflight, TriggerReceiptID: "reply-undo-test", CreatedAt: time.Now().UTC()}) + if err != nil { + t.Fatal(err) + } + paths := map[string]threadUndoPath{"tracked.txt": {beforeHash: fileedit.HashText("base\n"), beforeText: "base\n", afterHash: fileedit.HashText("changed\n"), afterText: "changed\n", modePath: "tracked.txt"}} + before := threadUndoSnapshot(observed, paths, false, "undo-before") + target := threadUndoSnapshot(observed, paths, true, "undo-target") + for _, snapshot := range []workspacecheckpoint.Snapshot{before, target} { + if _, _, err := f.state.CreateWorkspaceCheckpoint(t.Context(), snapshot); err != nil { + t.Fatal(err) + } + } + now := time.Now().UTC() + digest := threadUndoDigest(domain.InitialThreadID(f.run.ID), sourceID) + tx := workspacecheckpoint.Transaction{ID: "undo-recovery-journal", ProtocolVersion: workspacecheckpoint.ProtocolVersion, OperationKeyDigest: digest, + RequestFingerprint: runmutation.Fingerprint("reply-undo-recovery"), RunID: f.run.ID, WorkspaceID: w.WorkspaceID, Kind: workspacecheckpoint.TransactionUndo, + TriggerReceiptID: workspacecheckpoint.ThreadUndoReceiptPrefix + sourceID, BeforeCheckpointID: before.Checkpoint.ID, TargetCheckpointID: target.Checkpoint.ID, + Status: workspacecheckpoint.TransactionPrepared, RecoveryLevel: workspacecheckpoint.RecoveryComplete, ConflictJSON: "[]", CreatedAt: now, UpdatedAt: now} + tx, _, err = f.state.CreateWorkspaceCheckpointTransaction(t.Context(), tx) + if err != nil { + t.Fatal(err) + } + if _, claimed, err := f.state.ClaimThreadChangeUndo(t.Context(), tx); err != nil || !claimed { + t.Fatalf("initial claim: %v %v", claimed, err) + } + if _, claimed, err := f.state.ClaimThreadChangeUndo(t.Context(), tx); err != nil || claimed { + t.Fatalf("stale prepared claim entered IO: %v %v", claimed, err) + } + if err := f.state.CheckThreadChangeUndoIdle(t.Context(), domain.InitialThreadID(f.run.ID), f.run.ID, ""); err == nil { + t.Fatal("open undo did not reserve workspace") + } + // Recovery observes an interrupted process; user content is retained. + writeDrydockTestFile(t, filepath.Join(w.Path, "tracked.txt"), "user work after crash\n") + count, err := f.service.ReconcileWorkspaceCheckpoints(t.Context()) + if err != nil || count != 1 { + t.Fatalf("Drydock recovery: count=%d err=%v", count, err) + } + journal, found, err := f.state.GetWorkspaceCheckpointTransaction(t.Context(), tx.ID) + if err != nil || !found || journal.Status != workspacecheckpoint.TransactionInterrupted { + t.Fatalf("journal: %+v %v", journal, err) + } + after, err := f.state.GetWorkspaceCheckpointSnapshot(t.Context(), journal.AfterCheckpointID) + if err != nil || len(after.Entries) != 1 || after.Entries[0].WorktreeSHA256 != fileedit.HashText("user work after crash\n") { + t.Fatalf("selected recovery state: %+v %v", after, err) + } + current, found, err := f.state.GetRunFileDrydock(t.Context(), f.run.ID) + if err != nil || !found || current.LastCheckpointID != w.LastCheckpointID { + t.Fatalf("whole Drydock cursor moved: %+v %v", current, err) + } + if err := f.state.CheckThreadChangeUndoIdle(t.Context(), domain.InitialThreadID(f.run.ID), f.run.ID, ""); err != nil { + t.Fatalf("recovery retained reservation: %v", err) + } +} + +func TestThreadChangeUndoRejectsOmittedLivePath(t *testing.T) { + root := t.TempDir() + path := filepath.Join(root, "unobserved.txt") + if err := os.WriteFile(path, []byte("live\n"), 0600); err != nil { + t.Fatal(err) + } + paths := map[string]threadUndoPath{"unobserved.txt": {}} + if err := validateThreadUndoObservation(root, workspacecheckpoint.Snapshot{}, paths); err == nil { + t.Fatal("truncated capture treated live path as missing") + } +} diff --git a/internal/application/thread_change_undo_test.go b/internal/application/thread_change_undo_test.go new file mode 100644 index 000000000..aaee403b5 --- /dev/null +++ b/internal/application/thread_change_undo_test.go @@ -0,0 +1,224 @@ +package application_test + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/application" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/store" + "cyberagent-workbench/internal/workspacecheckpoint" +) + +func undoThreadReply(t *testing.T, st application.ThreadChangeUndoStore, run domain.Run, eventID string, confirm bool) application.ThreadChangeUndoResult { + t.Helper() + r, err := application.NewThreadChangeUndoService(st, domain.ExecutionPermissionRuntimeCapabilities{OperatorApprovalEnabled: true}).Undo(t.Context(), domain.InitialThreadID(run.ID), application.ThreadChangeUndoRequest{Version: application.ThreadChangeUndoVersion, SourceEventID: eventID, Confirm: confirm}) + if err != nil { + t.Fatal(err) + } + return r +} + +func TestThreadChangeUndoRestoresReplyIntervalAndDurablyReplays(t *testing.T) { + root := t.TempDir() + st, run := threadReviewFixture(t, root) + for path, text := range map[string]string{"app.txt": "before\n", "old.txt": "move me\n", "deleted.txt": "deleted\n", "outside.txt": "untouched\n"} { + if err := os.WriteFile(filepath.Join(root, path), []byte(text), 0600); err != nil { + t.Fatal(err) + } + } + applyThreadChange(t, st, run, root, "app.txt", "replace", "", "middle\n") + applyThreadChange(t, st, run, root, "app.txt", "replace", "", "after\n") + applyThreadChange(t, st, run, root, "old.txt", "move", "intermediate.txt", "") + applyThreadChange(t, st, run, root, "intermediate.txt", "move", "new.txt", "") + applyThreadChange(t, st, run, root, "new.txt", "replace", "", "changed after move\n") + deleteBefore, err := workspacecheckpoint.Capture(t.Context(), workspacecheckpoint.CaptureRequest{ID: "delete-before", RunID: run.ID, MissionID: run.MissionID, SessionID: run.SessionID, + WorkspaceID: "review-workspace", WorkspaceRoot: root, Trigger: workspacecheckpoint.TriggerFileTool, Phase: workspacecheckpoint.PhaseBefore, TriggerReceiptID: "pending-edit", CreatedAt: time.Now().UTC()}) + if err != nil { + t.Fatal(err) + } + deleted := applyThreadChange(t, st, run, root, "deleted.txt", "delete", "", "") + deleteBefore.Checkpoint.TriggerReceiptID = deleted.ID + if _, _, err := st.CreateWorkspaceCheckpoint(t.Context(), deleteBefore); err != nil { + t.Fatal(err) + } + applyThreadChange(t, st, run, root, "created.txt", "create", "", "new\n") + reply := saveThreadChangeReply(t, st, run, "done") + if err := os.WriteFile(filepath.Join(root, "outside.txt"), []byte("external change survives\n"), 0600); err != nil { + t.Fatal(err) + } + preview := undoThreadReply(t, st, run, reply.EventID, false) + if preview.State != "available" { + t.Fatalf("preview: %+v", preview) + } + if text, _ := os.ReadFile(filepath.Join(root, "app.txt")); string(text) != "after\n" { + t.Fatal("preview wrote files") + } + result := undoThreadReply(t, st, run, reply.EventID, true) + if result.State != "completed" || result.Replayed || len(result.RemainingPaths) != 0 { + t.Fatalf("undo: %+v", result) + } + for path, want := range map[string]string{"app.txt": "before\n", "old.txt": "move me\n", "deleted.txt": "deleted\n", "outside.txt": "external change survives\n"} { + text, err := os.ReadFile(filepath.Join(root, path)) + if err != nil || string(text) != want { + t.Fatalf("%s: %q %v", path, text, err) + } + } + for _, path := range []string{"new.txt", "intermediate.txt", "created.txt"} { + if _, err := os.Stat(filepath.Join(root, path)); !os.IsNotExist(err) { + t.Fatalf("%s still exists: %v", path, err) + } + } + if err := os.WriteFile(filepath.Join(root, "app.txt"), []byte("later user work\n"), 0600); err != nil { + t.Fatal(err) + } + replay := undoThreadReply(t, st, run, reply.EventID, true) + if replay.State != "completed" || !replay.Replayed { + t.Fatalf("replay: %+v", replay) + } + if text, _ := os.ReadFile(filepath.Join(root, "app.txt")); string(text) != "later user work\n" { + t.Fatal("replay overwrote later work") + } + if checkpoint, found, err := st.GetWorkspaceCheckpointRunState(t.Context(), run.ID); err != nil || found { + t.Fatalf("reply undo changed whole workspace cursor: %+v %v %v", checkpoint, found, err) + } + checkpoints, err := application.NewWorkspaceCheckpointService(st, domain.ExecutionPermissionRuntimeCapabilities{OperatorApprovalEnabled: true}) + if err != nil { + t.Fatal(err) + } + if _, err = checkpoints.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + timeline, err := checkpoints.Timeline(t.Context(), run.ID, 100) + if err != nil { + t.Fatal(err) + } + for _, c := range timeline.Checkpoints { + if workspacecheckpoint.IsThreadUndoCheckpoint(c) { + t.Fatal("scoped undo snapshot leaked into whole-workspace timeline") + } + } + transactions, err := st.ListWorkspaceCheckpointTransactions(t.Context(), run.ID, 100) + if err != nil { + t.Fatal(err) + } + for _, transaction := range transactions { + if workspacecheckpoint.IsThreadUndo(transaction) { + _, err = checkpoints.Restore(t.Context(), application.WorkspaceRestoreRequest{RunID: run.ID, TargetCheckpointID: transaction.TargetCheckpointID, Kind: workspacecheckpoint.TransactionRewind}) + if apperror.CodeOf(err) != apperror.CodeFailedPrecondition { + t.Fatalf("scoped target accepted by whole restore: %v", err) + } + } + } +} + +func TestThreadChangeUndoRejectsLaterWritesAndBrokenHistory(t *testing.T) { + root := t.TempDir() + st, run := threadReviewFixture(t, root) + if err := os.WriteFile(filepath.Join(root, "app.txt"), []byte("before\n"), 0600); err != nil { + t.Fatal(err) + } + applyThreadChange(t, st, run, root, "app.txt", "replace", "", "after\n") + reply := saveThreadChangeReply(t, st, run, "first") + applyThreadChange(t, st, run, root, "app.txt", "replace", "", "next turn\n") + saveThreadChangeReply(t, st, run, "second") + r := undoThreadReply(t, st, run, reply.EventID, true) + if r.State != "conflict" || len(r.Conflicts) != 1 || r.Conflicts[0].Path != "app.txt" { + t.Fatalf("conflict: %+v", r) + } + if text, _ := os.ReadFile(filepath.Join(root, "app.txt")); string(text) != "next turn\n" { + t.Fatal("later reply was overwritten") + } + if err := os.WriteFile(filepath.Join(root, "app.txt"), []byte("external gap\n"), 0600); err != nil { + t.Fatal(err) + } + applyThreadChange(t, st, run, root, "app.txt", "replace", "", "after gap\n") + if err := os.WriteFile(filepath.Join(root, "app.txt"), []byte("second gap\n"), 0600); err != nil { + t.Fatal(err) + } + applyThreadChange(t, st, run, root, "app.txt", "replace", "", "last\n") + gap := saveThreadChangeReply(t, st, run, "gap") + if r := undoThreadReply(t, st, run, gap.EventID, true); r.State != "unavailable" || r.Conflicts[0].Reason != "incomplete_history" { + t.Fatalf("gap: %+v", r) + } +} + +type threadUndoConcurrentWriter struct { + *store.SQLiteStore + root string + transaction workspacecheckpoint.Transaction +} + +func (s *threadUndoConcurrentWriter) ClaimThreadChangeUndo(ctx context.Context, t workspacecheckpoint.Transaction) (workspacecheckpoint.Transaction, bool, error) { + result, claimed, err := s.SQLiteStore.ClaimThreadChangeUndo(ctx, t) + if err == nil && claimed { + s.transaction = result + err = os.WriteFile(filepath.Join(s.root, "b.txt"), []byte("concurrent user content\n"), 0600) + } + return result, claimed, err +} + +func TestThreadChangeUndoPersistsPartialResultWithoutReapplying(t *testing.T) { + root := t.TempDir() + st, run := threadReviewFixture(t, root) + for _, path := range []string{"a.txt", "b.txt"} { + if err := os.WriteFile(filepath.Join(root, path), []byte("before\n"), 0600); err != nil { + t.Fatal(err) + } + applyThreadChange(t, st, run, root, path, "replace", "", "after\n") + } + reply := saveThreadChangeReply(t, st, run, "done") + wrapped := &threadUndoConcurrentWriter{SQLiteStore: st, root: root} + r := undoThreadReply(t, wrapped, run, reply.EventID, true) + if r.State != "partial" || len(r.RestoredPaths) != 1 || r.RestoredPaths[0] != "a.txt" || len(r.RemainingPaths) != 1 || r.RemainingPaths[0] != "b.txt" { + t.Fatalf("partial: %+v", r) + } + if err := os.WriteFile(filepath.Join(root, "a.txt"), []byte("later edit\n"), 0600); err != nil { + t.Fatal(err) + } + replay := undoThreadReply(t, st, run, reply.EventID, true) + if replay.State != "partial" || !replay.Replayed { + t.Fatalf("partial replay: %+v", replay) + } + if text, _ := os.ReadFile(filepath.Join(root, "a.txt")); string(text) != "later edit\n" { + t.Fatal("partial receipt reapplied files") + } + if err := st.CheckThreadChangeUndoIdle(t.Context(), domain.InitialThreadID(run.ID), run.ID, ""); err != nil { + t.Fatalf("terminal partial retained reservation: %v", err) + } +} + +func TestThreadChangeUndoExecutionLeaseBlocksMutation(t *testing.T) { + root := t.TempDir() + st, run := threadReviewFixture(t, root) + applyThreadChange(t, st, run, root, "app.txt", "create", "", "after\n") + reply := saveThreadChangeReply(t, st, run, "done") + _, err := st.AcquireRunExecutionLease(t.Context(), domain.AcquireRunExecutionLeaseRequest{RunID: run.ID, OwnerID: "active-model", TTL: time.Minute}) + if err != nil { + t.Fatal(err) + } + if r := undoThreadReply(t, st, run, reply.EventID, true); r.State != "unavailable" || r.Conflicts[0].Reason != "run_busy" { + t.Fatalf("active lease: %+v", r) + } +} + +func TestThreadChangeUndoRejectsDeletedFileWithoutOriginalMode(t *testing.T) { + root := t.TempDir() + st, run := threadReviewFixture(t, root) + if err := os.WriteFile(filepath.Join(root, "deleted.txt"), []byte("original\n"), 0600); err != nil { + t.Fatal(err) + } + applyThreadChange(t, st, run, root, "deleted.txt", "delete", "", "") + reply := saveThreadChangeReply(t, st, run, "deleted") + r := undoThreadReply(t, st, run, reply.EventID, true) + if r.State != "unavailable" || r.Conflicts[0].Reason != "incomplete_history" { + t.Fatalf("unknown original mode: %+v", r) + } + if _, err := os.Stat(filepath.Join(root, "deleted.txt")); !os.IsNotExist(err) { + t.Fatal("unknown mode was guessed") + } +} diff --git a/internal/application/thread_changes.go b/internal/application/thread_changes.go index b0a29d51e..f89171617 100644 --- a/internal/application/thread_changes.go +++ b/internal/application/thread_changes.go @@ -21,18 +21,19 @@ const ( ) type ThreadChangeSummary struct { - Version string `json:"version"` - ThreadID string `json:"thread_id"` - SourceEventID string `json:"source_event_id"` - RunID string `json:"run_id"` - ChangeScope string `json:"change_scope"` - Files []ThreadChangeFile `json:"files"` - FileCount int `json:"file_count"` - Additions int `json:"additions"` - Deletions int `json:"deletions"` - CountsAvailable bool `json:"counts_available"` - Partial bool `json:"partial"` - Reasons []string `json:"reasons"` + Version string `json:"version"` + ThreadID string `json:"thread_id"` + SourceEventID string `json:"source_event_id"` + RunID string `json:"run_id"` + ChangeScope string `json:"change_scope"` + Files []ThreadChangeFile `json:"files"` + FileCount int `json:"file_count"` + Additions int `json:"additions"` + Deletions int `json:"deletions"` + CountsAvailable bool `json:"counts_available"` + Partial bool `json:"partial"` + Reasons []string `json:"reasons"` + Undo *ThreadChangeUndoResult `json:"undo,omitempty"` } type ThreadChangeFile struct { diff --git a/internal/application/workspace_checkpoints.go b/internal/application/workspace_checkpoints.go index 54df8353d..b362d7c46 100644 --- a/internal/application/workspace_checkpoints.go +++ b/internal/application/workspace_checkpoints.go @@ -690,6 +690,19 @@ func (s *WorkspaceCheckpointService) Timeline(ctx context.Context, runID string, if err != nil { return WorkspaceCheckpointTimeline{}, apperror.Normalize(err) } + visibleCheckpoints := make([]workspacecheckpoint.Checkpoint, 0, len(checkpoints)) + for _, checkpoint := range checkpoints { + if !workspacecheckpoint.IsThreadUndoCheckpoint(checkpoint) { + visibleCheckpoints = append(visibleCheckpoints, checkpoint) + } + } + visibleTransactions := make([]workspacecheckpoint.Transaction, 0, len(transactions)) + for _, transaction := range transactions { + if !workspacecheckpoint.IsThreadUndo(transaction) { + visibleTransactions = append(visibleTransactions, transaction) + } + } + checkpoints, transactions = visibleCheckpoints, visibleTransactions usage, err := s.store.WorkspaceCheckpointStorageUsage(ctx) if err != nil { return WorkspaceCheckpointTimeline{}, apperror.Normalize(err) @@ -755,6 +768,9 @@ func (s *WorkspaceCheckpointService) Restore(ctx context.Context, if err != nil { return WorkspaceRestoreResult{}, apperror.Normalize(err) } + if workspacecheckpoint.IsThreadUndoCheckpoint(target.Checkpoint) { + return WorkspaceRestoreResult{}, apperror.New(apperror.CodeFailedPrecondition, "selected reply paths cannot be used as a whole Workspace checkpoint") + } if target.Checkpoint.RunID != binding.run.ID || target.Checkpoint.WorkspaceID != binding.workspace.ID { return WorkspaceRestoreResult{}, apperror.New(apperror.CodeNotFound, @@ -1284,6 +1300,13 @@ func (s *WorkspaceCheckpointService) reconcile(ctx context.Context, return reconciled, apperror.New(apperror.CodeConflict, "checkpoint recovery cannot change the recorded workspace") } + if workspacecheckpoint.IsThreadUndo(transaction) { + if err := scoped.reconcileThreadChangeUndo(ctx, binding, transaction); err != nil { + return reconciled, err + } + reconciled++ + continue + } before, getErr := scoped.store.GetWorkspaceCheckpointSnapshot(ctx, transaction.BeforeCheckpointID) if getErr != nil { diff --git a/internal/httpapi/openapi.go b/internal/httpapi/openapi.go index 3684167d7..c658ea171 100644 --- a/internal/httpapi/openapi.go +++ b/internal/httpapi/openapi.go @@ -1078,6 +1078,11 @@ func openAPIOperationSpecs() []openAPIOperationSpec { DataType: reflect.TypeOf(ThreadTranscriptItemView{}), Collection: true, Paginated: true, NotFound: true, Parameters: append([]openAPIParameter{threadID}, threadTranscriptPaginationParameters()...)}, + {Path: ThreadChangesUndoPathTemplate, Method: http.MethodPost, OperationID: "undoThreadChanges", + Summary: "Preview or undo recorded changes for one Thread reply", Tag: "Control", + Description: "With confirm=false, checks current authority and conflicts without writing. With confirm=true, restores only receipt-proven paths to their state before this reply interval. One durable operation is bound to the exact Thread and source event; completed and partial results replay without applying again. Requires an idle physical working directory and current control authorization.", + DataType: reflect.TypeOf(application.ThreadChangeUndoResult{}), RequestType: reflect.TypeOf(application.ThreadChangeUndoRequest{}), Control: true, + NotFound: true, Parameters: []openAPIParameter{threadID}, SuccessStatus: http.StatusOK}, {Path: ThreadChangesPathTemplate, OperationID: "getThreadChanges", Summary: "Read recorded file changes before one Thread reply", Tag: "Threads", Description: "Read-only net change summary for applied FileEdit records since the previous durable assistant reply in the same Run. The source event must be an assistant Session message in this Thread. Repeated edits are combined, unapplied proposals are excluded, and current workspace files are never read. Redacted or discontinuous histories omit counts and diff bodies; bounded diff omission is explicit.", diff --git a/internal/httpapi/openapi_test.go b/internal/httpapi/openapi_test.go index e0ef9c13c..911956004 100644 --- a/internal/httpapi/openapi_test.go +++ b/internal/httpapi/openapi_test.go @@ -1304,7 +1304,7 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { } else if spec.Path == FileEditProposalRecoveryPathTemplate { requestPath = strings.ReplaceAll(spec.Path, "{run_id}", fixture.run.ID) requestPath = strings.ReplaceAll(requestPath, "{edit_id}", recoveryProposal.Edit.ID) - } else if spec.OperationID == "branchThread" || spec.OperationID == "getThreadChanges" { + } else if spec.OperationID == "branchThread" || spec.OperationID == "getThreadChanges" || spec.OperationID == "undoThreadChanges" { requestPath = strings.ReplaceAll(spec.Path, "{thread_id}", domain.InitialThreadID(branchSourceRun.ID)) } else if spec.OperationID == "deleteThread" { requestPath = strings.ReplaceAll(spec.Path, "{thread_id}", deleteThread.ID) @@ -1453,6 +1453,8 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { t.Fatal(err) } body = string(encoded) + } else if spec.OperationID == "undoThreadChanges" { + body = `{"version":"thread_change_undo.v1","source_event_id":"` + branchSourceEventID + `","confirm":false}` } else if spec.OperationID == "startRunWorkflow" { body = `{"version":"workflow_start.v1","definition_id":"` + workflowDefinition.ID + `","inputs":{},"confirm_start":true}` } else if strings.HasPrefix(spec.Path, "/api/v1/runs/{run_id}/workflows/{execution_id}/") { diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 77c006e54..36567b923 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -1293,6 +1293,10 @@ func (a *API) ServeHTTP(writer http.ResponseWriter, request *http.Request) { a.serveThreadContextCompactionRecovery(tracked, request, requestID, threadID, operationID) return } + if threadID, matched := matchThreadChangesUndoPath(request.URL.Path); matched { + a.serveThreadChangesUndo(tracked, request, requestID, threadID) + return + } if threadID, action, matched := matchThreadMutationPath(request.URL.Path); matched && request.Method != http.MethodGet { a.serveThreadMutationControl(tracked, request, requestID, threadID, action) diff --git a/internal/httpapi/thread_change_undo_test.go b/internal/httpapi/thread_change_undo_test.go new file mode 100644 index 000000000..a1f349ff2 --- /dev/null +++ b/internal/httpapi/thread_change_undo_test.go @@ -0,0 +1,91 @@ +package httpapi + +import ( + "fmt" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + + "cyberagent-workbench/internal/application" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/fileedit" + "cyberagent-workbench/internal/session" + "cyberagent-workbench/internal/store" +) + +func TestThreadChangeUndoHTTPControlPreviewApplyAndRefresh(t *testing.T) { + f := newAPIFixture(t) + f.api.workspaceCheckpointControlEnabled = true + f.api.executionPermissionCapabilities.OperatorApprovalEnabled = true + f.workspace = store.WorkspaceRecord{ID: "undo-http-workspace", Name: "undo-http", RootPath: t.TempDir()} + if err := f.store.SaveWorkspace(t.Context(), f.workspace); err != nil { + t.Fatal(err) + } + _, run, err := application.NewRunService(f.store).Create(t.Context(), application.CreateRunRequest{Goal: "undo reply", Profile: "code", Surface: "code", Phase: "deliver", WorkspaceID: f.workspace.ID, Interactive: true}) + if err != nil { + t.Fatal(err) + } + f.run = run + manager := fileedit.NewManager(f.store) + e, err := manager.Propose(t.Context(), fileedit.Proposal{SessionID: f.run.SessionID, WorkspaceID: f.workspace.ID, WorkspaceRoot: f.workspace.RootPath, Path: "reply.txt", Operation: "create", ExpectedOriginalHash: "missing", ProposedText: "reply change\n"}) + if err != nil { + t.Fatal(err) + } + if _, err := application.NewFileEditReviewService(f.store).Review(t.Context(), application.ReviewFileEditRequest{Version: application.FileEditReviewProtocolVersion, RunID: f.run.ID, EditID: e.ID, Action: application.FileEditApproveIntent}); err != nil { + t.Fatal(err) + } + if _, err := manager.Approve(t.Context(), e.ID, f.workspace.RootPath); err != nil { + t.Fatal(err) + } + message, err := f.store.SaveSessionMessage(t.Context(), session.NewMessage(f.run.SessionID, "assistant", "done")) + if err != nil { + t.Fatal(err) + } + facts, err := f.store.ListRunEvents(t.Context(), f.run.ID) + if err != nil { + t.Fatal(err) + } + sourceID := "" + for _, fact := range facts { + if fact.Type == events.SessionMessageEvent && fact.SubjectID == fmt.Sprint(message.ID) { + sourceID = fact.EventID + } + } + threadID := domain.InitialThreadID(f.run.ID) + path := ThreadCollectionPath + "/" + threadID + "/changes/undo" + body := fmt.Sprintf(`{"version":"thread_change_undo.v1","source_event_id":%q,"confirm":true}`, sourceID) + denied := performSessionMessageRequest(t, f.api, http.MethodPost, path, testAccessToken, "", "", strings.NewReader(body)) + assertAPIError(t, denied, http.StatusUnauthorized, "POLICY_DENIED") + previewBody := strings.Replace(body, `"confirm":true`, `"confirm":false`, 1) + preview := performControlMethodPathRequest(t, f.api, http.MethodPost, path, "fixed-reply-key", strings.NewReader(previewBody)) + var result application.ThreadChangeUndoResult + decodeDataStatus(t, preview, http.StatusOK, &result) + if result.State != "available" { + t.Fatalf("preview: %+v", result) + } + if _, err := os.Stat(filepath.Join(f.workspace.RootPath, "reply.txt")); err != nil { + t.Fatal("preview changed file") + } + confirmed := performControlMethodPathRequest(t, f.api, http.MethodPost, path, "fixed-reply-key", strings.NewReader(body)) + decodeDataStatus(t, confirmed, http.StatusOK, &result) + if result.State != "completed" { + t.Fatalf("undo: %+v", result) + } + if _, err := os.Stat(filepath.Join(f.workspace.RootPath, "reply.txt")); !os.IsNotExist(err) { + t.Fatalf("creation not undone: %v", err) + } + refresh := performSessionMessageRequest(t, f.api, http.MethodGet, ThreadCollectionPath+"/"+threadID+"/changes?source_event_id="+sourceID, testAccessToken, "", "", nil) + var summary application.ThreadChangeSummary + decodeDataStatus(t, refresh, http.StatusOK, &summary) + if summary.Undo == nil || summary.Undo.State != "completed" || !summary.Undo.Replayed { + t.Fatalf("refresh lost receipt: %+v", summary.Undo) + } + foreign := performControlMethodPathRequest(t, f.api, http.MethodPost, ThreadCollectionPath+"/foreign-thread/changes/undo", "fixed-reply-key", strings.NewReader(body)) + assertAPIError(t, foreign, http.StatusNotFound, "NOT_FOUND") + f.api.workspaceCheckpointControlEnabled = false + disabled := performControlMethodPathRequest(t, f.api, http.MethodPost, path, "fixed-reply-key", strings.NewReader(body)) + assertAPIError(t, disabled, http.StatusNotFound, "NOT_FOUND") +} diff --git a/internal/httpapi/thread_changes.go b/internal/httpapi/thread_changes.go index 94fbbf7ae..1323c45cc 100644 --- a/internal/httpapi/thread_changes.go +++ b/internal/httpapi/thread_changes.go @@ -2,12 +2,14 @@ package httpapi import ( "net/http" + "strings" "cyberagent-workbench/internal/apperror" "cyberagent-workbench/internal/application" ) const ThreadChangesPathTemplate = "/api/v1/threads/{thread_id}/changes" +const ThreadChangesUndoPathTemplate = "/api/v1/threads/{thread_id}/changes/undo" func (a *API) threadChanges(request *http.Request, threadID string) (any, *Page, error) { query := request.URL.Query() @@ -22,5 +24,60 @@ func (a *API) threadChanges(request *http.Request, threadID string) (any, *Page, return nil, nil, apperror.New(apperror.CodeUnavailable, "Thread change summary is unavailable") } result, err := application.NewThreadChangeService(reader).Summary(request.Context(), threadID, query.Get("source_event_id")) + if err == nil { + if store, ok := a.store.(application.ThreadChangeUndoStore); ok { + status, statusErr := application.NewThreadChangeUndoService(store, a.executionPermissionCapabilities).Status(request.Context(), result) + if statusErr != nil { + return nil, nil, statusErr + } + result.Undo = &status + } + } return result, nil, err } + +func matchThreadChangesUndoPath(path string) (string, bool) { + parts := strings.Split(strings.TrimPrefix(path, ThreadCollectionPath+"/"), "/") + if strings.HasPrefix(path, ThreadCollectionPath+"/") && len(parts) == 3 && parts[0] != "" && parts[1] == "changes" && parts[2] == "undo" { + return parts[0], true + } + return "", false +} + +func (a *API) serveThreadChangesUndo(writer http.ResponseWriter, request *http.Request, requestID, threadID string) { + if !a.workspaceCheckpointControlEnabled { + a.writeError(writer, requestID, apperror.New(apperror.CodeNotFound, "Thread change undo is unavailable"), http.StatusNotFound) + return + } + if !a.authorizeThreadControl(writer, request, requestID) { + return + } + if request.Method != http.MethodPost { + writer.Header().Set("Allow", http.MethodPost) + a.writeError(writer, requestID, apperror.New(apperror.CodeInvalidArgument, "Thread change undo only supports POST"), http.StatusMethodNotAllowed) + return + } + if err := validatePathIdentity(threadID); err != nil { + a.writeError(writer, requestID, err, 0) + return + } + if err := rejectQuery(request.URL.Query()); err != nil { + a.writeError(writer, requestID, err, 0) + return + } + var view application.ThreadChangeUndoRequest + if !a.decodeThreadControlBody(writer, request, requestID, "Thread change undo", &view) { + return + } + store, ok := a.store.(application.ThreadChangeUndoStore) + if !ok { + a.writeError(writer, requestID, apperror.New(apperror.CodeUnavailable, "Thread change undo storage is unavailable"), 0) + return + } + result, err := application.NewThreadChangeUndoService(store, a.executionPermissionCapabilities).WithDrydocks(a.fileWorkspaceDrydocks).Undo(request.Context(), threadID, view) + if err != nil { + a.writeError(writer, requestID, err, 0) + return + } + a.writeSuccessStatus(writer, requestID, result, nil, http.StatusOK) +} diff --git a/internal/store/thread_change_undo.go b/internal/store/thread_change_undo.go new file mode 100644 index 000000000..77047e938 --- /dev/null +++ b/internal/store/thread_change_undo.go @@ -0,0 +1,86 @@ +package store + +import ( + "context" + "database/sql" + "strings" + "time" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/workspacecheckpoint" +) + +func (s *SQLiteStore) ClaimThreadChangeUndo(ctx context.Context, t workspacecheckpoint.Transaction) (workspacecheckpoint.Transaction, bool, error) { + if !workspacecheckpoint.IsThreadUndo(t) || t.Status != workspacecheckpoint.TransactionPrepared { + return t, false, apperror.New(apperror.CodeConflict, "Thread undo claim is invalid") + } + now := time.Now().UTC() + result, err := s.db.ExecContext(ctx, `UPDATE workspace_checkpoint_transactions SET status='applying',updated_at=? + WHERE id=? AND operation_key_digest=? AND request_fingerprint=? AND status='prepared' AND updated_at=?`, ts(now), t.ID, t.OperationKeyDigest, t.RequestFingerprint, ts(t.UpdatedAt)) + if err != nil { + return t, false, err + } + n, err := result.RowsAffected() + if err != nil { + return t, false, err + } + current, _, err := s.GetWorkspaceCheckpointTransaction(ctx, t.ID) + return current, n == 1, err +} + +func (s *SQLiteStore) CheckThreadChangeUndoIdle(ctx context.Context, threadID, runID, ownTransaction string) error { + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) + if err != nil { + return err + } + defer tx.Rollback() + return requireThreadChangeUndoIdleTx(ctx, tx, threadID, runID, ownTransaction) +} + +// The same check is repeated by journal insertion in its write transaction. +// Reservations fence execution on every Run sharing the physical Workspace. +func requireThreadChangeUndoIdleTx(ctx context.Context, tx *sql.Tx, threadID, runID, ownTransaction string) error { + var workspaceID string + if err := tx.QueryRowContext(ctx, `SELECT COALESCE(owner.workspace_id,mission.workspace_id) + FROM threads thread JOIN runs run ON run.id=thread.last_run_id + JOIN missions mission ON mission.id=run.mission_id + LEFT JOIN run_file_drydock_bindings owner ON owner.run_id=run.id + WHERE thread.id=? AND thread.last_run_id=? AND thread.status='active'`, threadID, runID).Scan(&workspaceID); err != nil { + return err + } + var busy bool + if err := tx.QueryRowContext(ctx, `WITH shared AS ( + SELECT run.id FROM runs run JOIN missions mission ON mission.id=run.mission_id + LEFT JOIN run_file_drydock_bindings owner ON owner.run_id=run.id + WHERE COALESCE(owner.workspace_id,mission.workspace_id)=? + ) SELECT + EXISTS(SELECT 1 FROM run_execution_leases lease JOIN shared ON shared.id=lease.run_id + WHERE lease.status='active' AND julianday(lease.expires_at)>julianday(?)) + OR EXISTS(SELECT 1 FROM workspace_checkpoint_transactions WHERE workspace_id=? AND id<>? AND status IN ('prepared','applying')) + OR EXISTS(SELECT 1 FROM command_runtime_jobs job JOIN shared ON shared.id=job.run_id WHERE job.state IN ('prepared','running','stopping') OR job.tree_reaped<>1) + OR EXISTS(SELECT 1 FROM run_supervisor_tool_calls tool JOIN shared ON shared.id=tool.run_id WHERE tool.status IN ('pending','running')) + OR EXISTS(SELECT 1 FROM terminal_sessions terminal JOIN shared ON shared.id=terminal.run_id WHERE terminal.state IN ('starting','running')) + OR EXISTS(SELECT 1 FROM operator_steering_deliveries input JOIN shared ON shared.id=input.run_id WHERE input.status='prepared') + OR EXISTS(SELECT 1 FROM drydock_cleanup_operations cleanup JOIN drydock_workspaces d ON d.id=cleanup.drydock_id WHERE d.workspace_id=? AND cleanup.status='prepared')`, + workspaceID, ts(time.Now().UTC()), workspaceID, ownTransaction, workspaceID).Scan(&busy); err != nil { + return err + } + if busy { + return apperror.New(apperror.CodeConflict, "Thread change undo requires an idle working directory") + } + return requireQuiescentRunPauseTx(ctx, tx, runID) +} + +func requireThreadChangeUndoJournalTx(ctx context.Context, tx *sql.Tx, value workspacecheckpoint.Transaction) error { + var threadID string + sourceID := strings.TrimPrefix(value.TriggerReceiptID, workspacecheckpoint.ThreadUndoReceiptPrefix) + if err := tx.QueryRowContext(ctx, `SELECT thread.id FROM threads thread + JOIN thread_runs binding ON binding.thread_id=thread.id + JOIN run_events event ON event.run_id=binding.run_id + WHERE thread.last_run_id=? AND event.event_id=? AND event.type='session.message_created' + AND event.source='session_store' AND json_extract(event.payload_json,'$.role')='assistant' + AND json_extract(event.payload_json,'$.source_kind')='model_response'`, value.RunID, sourceID).Scan(&threadID); err != nil { + return err + } + return requireThreadChangeUndoIdleTx(ctx, tx, threadID, value.RunID, "") +} diff --git a/internal/store/workspace_checkpoints.go b/internal/store/workspace_checkpoints.go index 0499ec4d0..d48e880d3 100644 --- a/internal/store/workspace_checkpoints.go +++ b/internal/store/workspace_checkpoints.go @@ -327,7 +327,11 @@ func (s *SQLiteStore) CreateWorkspaceCheckpointTransaction(ctx context.Context, `SELECT `+workspaceCheckpointTransactionColumns+` FROM workspace_checkpoint_transactions WHERE operation_key_digest = ?`, value.OperationKeyDigest)) if lookupErr == nil { - if !sameWorkspaceCheckpointTransactionIntent(existing, value) { + sameIntent := sameWorkspaceCheckpointTransactionIntent(existing, value) + if workspacecheckpoint.IsThreadUndo(existing) && workspacecheckpoint.IsThreadUndo(value) { + sameIntent = existing.RunID == value.RunID && existing.WorkspaceID == value.WorkspaceID && existing.TriggerReceiptID == value.TriggerReceiptID && existing.RequestFingerprint == value.RequestFingerprint + } + if !sameIntent { return workspacecheckpoint.Transaction{}, false, apperror.New( apperror.CodeConflict, "workspace checkpoint operation key was reused") } @@ -339,18 +343,23 @@ func (s *SQLiteStore) CreateWorkspaceCheckpointTransaction(ctx context.Context, if !errors.Is(lookupErr, sql.ErrNoRows) { return workspacecheckpoint.Transaction{}, false, lookupErr } - var pausedAndActive bool - if err := tx.QueryRowContext(ctx, `SELECT EXISTS ( + if workspacecheckpoint.IsThreadUndo(value) { + if err := requireThreadChangeUndoJournalTx(ctx, tx, value); err != nil { + return workspacecheckpoint.Transaction{}, false, err + } + } else { + var pausedAndActive bool + if err := tx.QueryRowContext(ctx, `SELECT EXISTS ( SELECT 1 FROM runs run JOIN sessions session_record ON session_record.id = run.session_id WHERE run.id = ? AND run.status = 'paused' AND session_record.status = 'active' AND session_record.workspace_id = ?)`, value.RunID, value.WorkspaceID).Scan(&pausedAndActive); err != nil { - return workspacecheckpoint.Transaction{}, false, err - } - if !pausedAndActive { - if hasBindings, err := hasRunFileDrydockBindings(ctx, tx); err != nil { return workspacecheckpoint.Transaction{}, false, err - } else if hasBindings { - if err := tx.QueryRowContext(ctx, `SELECT EXISTS ( + } + if !pausedAndActive { + if hasBindings, err := hasRunFileDrydockBindings(ctx, tx); err != nil { + return workspacecheckpoint.Transaction{}, false, err + } else if hasBindings { + if err := tx.QueryRowContext(ctx, `SELECT EXISTS ( SELECT 1 FROM runs run JOIN sessions linked ON linked.id=run.session_id JOIN run_file_drydock_bindings owner ON owner.run_id=run.id AND owner.session_id=linked.id JOIN drydock_workspaces d ON d.id=owner.drydock_id @@ -358,12 +367,12 @@ func (s *SQLiteStore) CreateWorkspaceCheckpointTransaction(ctx context.Context, WHERE run.id=? AND run.status='paused' AND linked.status='active' AND linked.workspace_id=owner.source_workspace_id AND owner.workspace_id=? AND d.state<>'cleaned' AND (owner.thread_id='' OR thread.last_run_id=run.id))`, - value.RunID, value.WorkspaceID).Scan(&pausedAndActive); err != nil { - return workspacecheckpoint.Transaction{}, false, err - } - if pausedAndActive { - var busy bool - if err := tx.QueryRowContext(ctx, `SELECT EXISTS ( + value.RunID, value.WorkspaceID).Scan(&pausedAndActive); err != nil { + return workspacecheckpoint.Transaction{}, false, err + } + if pausedAndActive { + var busy bool + if err := tx.QueryRowContext(ctx, `SELECT EXISTS ( SELECT 1 FROM run_file_drydock_bindings current WHERE current.run_id=? AND current.workspace_id=? AND ( EXISTS(SELECT 1 FROM run_file_drydock_bindings other JOIN run_execution_leases lease ON lease.run_id=other.run_id @@ -372,22 +381,23 @@ func (s *SQLiteStore) CreateWorkspaceCheckpointTransaction(ctx context.Context, WHERE pending.workspace_id=current.workspace_id AND pending.status IN ('prepared','applying')) OR EXISTS(SELECT 1 FROM drydock_cleanup_operations cleanup WHERE cleanup.drydock_id=current.drydock_id AND cleanup.status='prepared')))`, - value.RunID, value.WorkspaceID, ts(time.Now().UTC())).Scan(&busy); err != nil { - return workspacecheckpoint.Transaction{}, false, err - } - if busy { - return workspacecheckpoint.Transaction{}, false, apperror.New(apperror.CodeConflict, - "Thread working directory has an unfinished execution, mutation, or cleanup") + value.RunID, value.WorkspaceID, ts(time.Now().UTC())).Scan(&busy); err != nil { + return workspacecheckpoint.Transaction{}, false, err + } + if busy { + return workspacecheckpoint.Transaction{}, false, apperror.New(apperror.CodeConflict, + "Thread working directory has an unfinished execution, mutation, or cleanup") + } } } } - } - if !pausedAndActive { - return workspacecheckpoint.Transaction{}, false, apperror.New(apperror.CodeFailedPrecondition, - "workspace restore preparation requires a paused Run and active Session") - } - if err := requireNoActiveRunControlLeaseTx(ctx, tx, value.RunID, time.Now().UTC()); err != nil { - return workspacecheckpoint.Transaction{}, false, err + if !pausedAndActive { + return workspacecheckpoint.Transaction{}, false, apperror.New(apperror.CodeFailedPrecondition, + "workspace restore preparation requires a paused Run and active Session") + } + if err := requireNoActiveRunControlLeaseTx(ctx, tx, value.RunID, time.Now().UTC()); err != nil { + return workspacecheckpoint.Transaction{}, false, err + } } } err = insertWorkspaceCheckpointTransactionTx(ctx, tx, value) @@ -439,11 +449,23 @@ func isWorkspaceRestoreTransaction(kind workspacecheckpoint.TransactionKind) boo // recovery finishes. Resume and lease acquisition check this in their own write // transactions, so neither can enter between restore preparation and file IO. func requireNoOpenWorkspaceRestoreTx(ctx context.Context, tx *sql.Tx, runID string) error { + // Historical stores use the mission workspace before physical owner bindings exist. + workspaceQuery := `SELECT mission.workspace_id FROM runs run + JOIN missions mission ON mission.id=run.mission_id WHERE run.id=?` + if hasBindings, err := hasRunFileDrydockBindings(ctx, tx); err != nil { + return err + } else if hasBindings { + workspaceQuery = `SELECT COALESCE(owner.workspace_id,mission.workspace_id) + FROM runs run JOIN missions mission ON mission.id=run.mission_id + LEFT JOIN run_file_drydock_bindings owner ON owner.run_id=run.id WHERE run.id=?` + } var pending bool if err := tx.QueryRowContext(ctx, `SELECT EXISTS ( - SELECT 1 FROM workspace_checkpoint_transactions WHERE run_id = ? + SELECT 1 FROM workspace_checkpoint_transactions pending + WHERE (pending.run_id = ? OR (pending.trigger_receipt_id LIKE 'thread-reply-undo:%' + AND pending.workspace_id=(`+workspaceQuery+`))) AND kind IN ('rewind', 'undo', 'redo', 'fork') AND status IN ('prepared', 'applying'))`, - runID).Scan(&pending); err != nil { + runID, runID).Scan(&pending); err != nil { return err } if pending { @@ -615,6 +637,7 @@ func (s *SQLiteStore) ListWorkspaceCheckpointTransactionsPendingCursor(ctx conte FROM workspace_checkpoint_transactions transaction_record WHERE transaction_record.status IN ('completed', 'failed', 'interrupted') AND transaction_record.kind != 'fork' + AND transaction_record.trigger_receipt_id NOT LIKE 'thread-reply-undo:%' AND transaction_record.after_checkpoint_id != '' AND (EXISTS (SELECT 1 FROM workspace_checkpoint_run_state state WHERE state.run_id = transaction_record.run_id diff --git a/internal/workspacecheckpoint/thread_undo.go b/internal/workspacecheckpoint/thread_undo.go new file mode 100644 index 000000000..d76f56a4f --- /dev/null +++ b/internal/workspacecheckpoint/thread_undo.go @@ -0,0 +1,14 @@ +package workspacecheckpoint + +import "strings" + +// Thread reply undo uses the existing restore journal, with a scoped manifest +// and its own recovery owner. It must never advance the whole-workspace cursor. +const ThreadUndoReceiptPrefix = "thread-reply-undo:" +const ThreadUndoSnapshotTitle = "Thread reply undo (selected paths)" + +func IsThreadUndo(t Transaction) bool { + return t.Kind == TransactionUndo && strings.HasPrefix(t.TriggerReceiptID, ThreadUndoReceiptPrefix) +} + +func IsThreadUndoCheckpoint(c Checkpoint) bool { return c.Title == ThreadUndoSnapshotTitle } diff --git a/protocols/registry.json b/protocols/registry.json index 31fa9200d..763514dce 100644 --- a/protocols/registry.json +++ b/protocols/registry.json @@ -1795,6 +1795,7 @@ "thread_application_services.v1", "thread_branch.v1", "thread_change_summary.v1", + "thread_change_undo.v1", "thread_command_service_policy.v1", "thread_context_compaction.v1", "thread_context_compaction_operations.v1", diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 5d0cd05b5..b6e9642fa 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -1,5 +1,5 @@ import { WorkflowClient } from "./workflows"; -import type { ThreadChangeSummary } from "./thread-changes"; +import type { ThreadChangeSummary, ThreadChangeUndo } from "./thread-changes"; import { creationBudgetMatches, normalizedTaskBudget, parseTaskConfiguration } from "./task-configuration"; import { parseTaskSkillCandidates, type TaskSkillCandidateScope } from "./task-skill-candidates"; import { validSkillSelectionRef } from "./project-skills"; @@ -8999,6 +8999,24 @@ export class APIClient { return value; } + async undoThreadChanges(threadID: string, runID: string, sourceEventID: string, + confirm: boolean, signal?: AbortSignal): Promise { + if (!this.hasWorkspaceCheckpointControl || !boundedIdentity(threadID) || + !boundedIdentity(runID) || !boundedIdentity(sourceEventID)) { + throw new Error("Workspace restore authority and an exact reply are required."); + } + const value = await this.sendControl(`/threads/${encodeURIComponent(threadID)}/changes/undo`, + { version: "thread_change_undo.v1", source_event_id: sourceEventID, confirm }, + `thread-change-undo-${sourceEventID}`, signal); + if (value.version !== "thread_change_undo.v1" || value.thread_id !== threadID || + value.run_id !== runID || value.source_event_id !== sourceEventID || + !["available", "unavailable", "conflict", "applying", "completed", "partial"].includes(value.state) || + !Array.isArray(value.conflicts) || !Array.isArray(value.restored_paths) || !Array.isArray(value.remaining_paths)) { + throw new Error("The undo result does not match this reply."); + } + return value; + } + async inspectThreadCreationRequest(workspaceID: string, operationKey: string, signal?: AbortSignal): Promise { if (!boundedIdentity(workspaceID) || workspaceID.trim() !== workspaceID) { diff --git a/web/src/api/schema.d.ts b/web/src/api/schema.d.ts index da3f56e1d..374a9bfe6 100644 --- a/web/src/api/schema.d.ts +++ b/web/src/api/schema.d.ts @@ -4868,6 +4868,26 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/threads/{thread_id}/changes/undo": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Preview or undo recorded changes for one Thread reply + * @description With confirm=false, checks current authority and conflicts without writing. With confirm=true, restores only receipt-proven paths to their state before this reply interval. One durable operation is bound to the exact Thread and source event; completed and partial results replay without applying again. Requires an idle physical working directory and current control authorization. + */ + post: operations["undoThreadChanges"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/threads/{thread_id}/command-service-policy": { parameters: { query?: never; @@ -14342,6 +14362,27 @@ export interface components { run_id: string; source_event_id: string; thread_id: string; + undo?: components["schemas"]["ThreadChangeUndoResult"]; + version: string; + }; + ThreadChangeUndoConflict: { + path?: string; + reason: string; + }; + ThreadChangeUndoRequest: { + confirm: boolean; + source_event_id: string; + version: string; + }; + ThreadChangeUndoResult: { + conflicts: components["schemas"]["ThreadChangeUndoConflict"][]; + remaining_paths: string[]; + replayed: boolean; + restored_paths: string[]; + run_id: string; + source_event_id: string; + state: string; + thread_id: string; version: string; }; ThreadCommandServicePolicy: { @@ -28244,6 +28285,51 @@ export interface operations { 504: components["responses"]["GatewayTimeout"]; }; }; + undoThreadChanges: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Thread identity */ + thread_id: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["ThreadChangeUndoRequest"]; + }; + }; + responses: { + /** @description Control request accepted or idempotently replayed */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + data: components["schemas"]["ThreadChangeUndoResult"]; + request_id: string; + /** @constant */ + version: "api.v1"; + }; + }; + }; + 400: components["responses"]["BadRequest"]; + 401: components["responses"]["Unauthorized"]; + 403: components["responses"]["Forbidden"]; + 404: components["responses"]["NotFound"]; + 409: components["responses"]["Conflict"]; + 412: components["responses"]["FailedPrecondition"]; + 413: components["responses"]["RequestEntityTooLarge"]; + 414: components["responses"]["RequestTooLarge"]; + 415: components["responses"]["UnsupportedMediaType"]; + 429: components["responses"]["ResourceExhausted"]; + 500: components["responses"]["InternalError"]; + 503: components["responses"]["Unavailable"]; + 504: components["responses"]["GatewayTimeout"]; + }; + }; getThreadCommandServicePolicy: { parameters: { query?: never; diff --git a/web/src/api/thread-changes.test.ts b/web/src/api/thread-changes.test.ts new file mode 100644 index 000000000..d74328640 --- /dev/null +++ b/web/src/api/thread-changes.test.ts @@ -0,0 +1,43 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { APIClient } from "./client"; + +afterEach(() => vi.unstubAllGlobals()); + +const receipt = { version: "thread_change_undo.v1", thread_id: "thread-1", run_id: "run-1", source_event_id: "reply-1", + state: "available", conflicts: [], restored_paths: [], remaining_paths: ["README.md"], replayed: false }; +function response(data: unknown) { + return new Response(JSON.stringify({ version: "api.v1", request_id: "test-undo", data }), + { status: 200, headers: { "Content-Type": "application/json" } }); +} + +it("uses control authorization and a stable reply key for preview, apply and retry", async () => { + const fetch = vi.fn().mockImplementation(async () => response(receipt)); + vi.stubGlobal("fetch", fetch); + const client = new APIClient("read-test", "/api/v1", "control-test", { workspaceCheckpointControlEnabled: true }); + await client.undoThreadChanges("thread-1", "run-1", "reply-1", false); + await client.undoThreadChanges("thread-1", "run-1", "reply-1", true); + await client.undoThreadChanges("thread-1", "run-1", "reply-1", true); + for (const [url, init] of fetch.mock.calls) { + expect(url).toBe("/api/v1/threads/thread-1/changes/undo"); + expect(init.method).toBe("POST"); + expect(init.headers.Authorization).toBe("Bearer control-test"); + expect(init.headers["Idempotency-Key"]).toBe("thread-change-undo-reply-1"); + expect(JSON.parse(init.body)).toMatchObject({ version: "thread_change_undo.v1", source_event_id: "reply-1" }); + } + expect(fetch.mock.calls.map(([, init]) => JSON.parse(init.body).confirm)).toEqual([false, true, true]); +}); + +it("rejects another reply's undo receipt", async () => { + const fetch = vi.fn().mockResolvedValue(response({ ...receipt, source_event_id: "other-reply", state: "completed" })); + vi.stubGlobal("fetch", fetch); + const client = new APIClient("read-test", "/api/v1", "control-test", { workspaceCheckpointControlEnabled: true }); + await expect(client.undoThreadChanges("thread-1", "run-1", "reply-1", true)).rejects.toThrow("does not match this reply"); +}); + +it("does not send undo requests without workspace restore authority", async () => { + const fetch = vi.fn(); + vi.stubGlobal("fetch", fetch); + const client = new APIClient("read-test", "/api/v1", "control-test", { workspaceCheckpointControlEnabled: false }); + await expect(client.undoThreadChanges("thread-1", "run-1", "reply-1", true)).rejects.toThrow("restore authority"); + expect(fetch).not.toHaveBeenCalled(); +}); diff --git a/web/src/api/thread-changes.ts b/web/src/api/thread-changes.ts index cc1834c3e..fdd04b72f 100644 --- a/web/src/api/thread-changes.ts +++ b/web/src/api/thread-changes.ts @@ -2,3 +2,4 @@ import type { components } from "./schema"; export type ThreadChangeSummary = components["schemas"]["ThreadChangeSummary"]; export type ThreadChangeFile = components["schemas"]["ThreadChangeFile"]; +export type ThreadChangeUndo = components["schemas"]["ThreadChangeUndoResult"]; diff --git a/web/src/v2/components/turn-change-undo.test.tsx b/web/src/v2/components/turn-change-undo.test.tsx new file mode 100644 index 000000000..2c8af54f8 --- /dev/null +++ b/web/src/v2/components/turn-change-undo.test.tsx @@ -0,0 +1,122 @@ +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { cleanup, render, screen, waitFor, within } from "@testing-library/react"; +import userEvent from "@testing-library/user-event"; +import { afterEach, expect, it, vi } from "vitest"; +import type { APIClient } from "../../api/client"; +import type { ThreadChangeSummary, ThreadChangeUndo } from "../../api/thread-changes"; +import { TurnChangeUndo } from "./turn-change-undo"; + +afterEach(cleanup); + +const summary: ThreadChangeSummary = { + version: "thread_change_summary.v1", thread_id: "thread-1", run_id: "run-1", source_event_id: "reply-1", + change_scope: "recorded_file_edits", file_count: 2, additions: 2, deletions: 2, + counts_available: true, partial: false, reasons: [], files: [], +}; +function result(overrides: Partial = {}): ThreadChangeUndo { + return { version: "thread_change_undo.v1", thread_id: "thread-1", run_id: "run-1", source_event_id: "reply-1", + state: "available", conflicts: [], restored_paths: [], remaining_paths: ["README.md", "app.ts"], replayed: false, ...overrides }; +} +function show(undoThreadChanges: ReturnType, undo?: ThreadChangeUndo) { + const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false }, mutations: { retry: false } } }); + const key = ["thread", "thread-1", "turn-changes", "reply-1"]; + queryClient.setQueryData(key, summary); + render( + + ); + return { queryClient, key }; +} + +it("previews only on demand and requires confirmation before undoing the exact reply", async () => { + const user = userEvent.setup(); + const api = vi.fn().mockResolvedValueOnce(result()).mockResolvedValueOnce(result({ state: "completed", restored_paths: ["README.md", "app.ts"], remaining_paths: [] })); + const { queryClient, key } = show(api); + expect(api).not.toHaveBeenCalled(); + await user.click(screen.getByRole("button", { name: "Undo this turn’s changes" })); + const confirm = await screen.findByRole("button", { name: "Confirm undo" }); + expect(api).toHaveBeenCalledExactlyOnceWith("thread-1", "run-1", "reply-1", false, expect.any(AbortSignal)); + await user.click(confirm); + await screen.findByRole("heading", { name: "This turn’s changes were undone" }); + expect(api).toHaveBeenLastCalledWith("thread-1", "run-1", "reply-1", true); + expect(queryClient.getQueryData(key)?.undo?.state).toBe("completed"); + await user.click(within(screen.getByRole("dialog")).getAllByRole("button", { name: "Close" })[1]); + const completed = screen.getByRole("button", { name: "This turn’s changes were undone" }); + expect(completed).toHaveAttribute("aria-disabled", "true"); + expect(completed).toHaveFocus(); + await user.click(completed); + expect(screen.queryByRole("dialog")).not.toBeInTheDocument(); + expect(api).toHaveBeenCalledTimes(2); +}); + +it("cancels without applying and returns keyboard focus to the card", async () => { + const user = userEvent.setup(); + const api = vi.fn().mockResolvedValue(result()); + show(api); + const trigger = screen.getByRole("button", { name: "Undo this turn’s changes" }); + await user.click(trigger); + await screen.findByRole("button", { name: "Confirm undo" }); + await user.keyboard("{Escape}"); + expect(screen.queryByRole("dialog")).not.toBeInTheDocument(); + expect(trigger).toHaveFocus(); + expect(api.mock.calls.every((call) => call[3] === false)).toBe(true); +}); + +it("shows the conflicting file and does not offer confirmation for later edits", async () => { + const user = userEvent.setup(); + const api = vi.fn().mockResolvedValue(result({ state: "conflict", conflicts: [{ path: "app.ts", reason: "files_changed" }] })); + show(api); + await user.click(screen.getByRole("button", { name: "Undo this turn’s changes" })); + await screen.findByText("app.ts"); + expect(screen.getByText("This file changed after this turn. Review its current content first.")).toBeInTheDocument(); + expect(screen.queryByRole("button", { name: "Confirm undo" })).not.toBeInTheDocument(); + expect(api.mock.calls.every((call) => call[3] === false)).toBe(true); +}); + +it("checks a lost response without applying the undo a second time", async () => { + const user = userEvent.setup(); + const api = vi.fn().mockResolvedValueOnce(result()).mockRejectedValueOnce(new Error("Connection lost")) + .mockResolvedValueOnce(result({ state: "completed", restored_paths: ["README.md", "app.ts"], remaining_paths: [], replayed: true })); + show(api); + await user.click(screen.getByRole("button", { name: "Undo this turn’s changes" })); + await user.click(await screen.findByRole("button", { name: "Confirm undo" })); + await screen.findByRole("alert"); + await user.click(screen.getByRole("button", { name: "Check again" })); + await screen.findByRole("heading", { name: "This turn’s changes were undone" }); + expect(api.mock.calls.map((call) => call[3])).toEqual([false, true, false]); +}); + +it("keeps completed undo state after reloading the summary", () => { + const api = vi.fn(); + show(api, result({ state: "completed", restored_paths: ["README.md", "app.ts"], remaining_paths: [] })); + expect(screen.getByRole("button", { name: "This turn’s changes were undone" })).toHaveAttribute("aria-disabled", "true"); + expect(api).not.toHaveBeenCalled(); +}); + +it("prefers a refreshed completed receipt over an earlier available preview", async () => { + const user = userEvent.setup(); + const api = vi.fn().mockResolvedValue(result()); + const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }); + const control = (value: ThreadChangeSummary) => + + ; + const view = render(control(summary)); + await user.click(screen.getByRole("button", { name: "Undo this turn’s changes" })); + await screen.findByRole("button", { name: "Confirm undo" }); + await user.click(screen.getByRole("button", { name: "Cancel" })); + view.rerender(control({ ...summary, undo: result({ state: "completed" }) })); + expect(screen.getByRole("button", { name: "This turn’s changes were undone" })).toHaveAttribute("aria-disabled", "true"); +}); + +it("reports partial progress without claiming the entire turn was undone", async () => { + const user = userEvent.setup(); + const api = vi.fn().mockResolvedValueOnce(result()).mockResolvedValueOnce(result({ state: "partial", + restored_paths: ["README.md"], remaining_paths: ["app.ts"], conflicts: [{ path: "app.ts", reason: "restore_failed" }] })); + show(api); + await user.click(screen.getByRole("button", { name: "Undo this turn’s changes" })); + await user.click(await screen.findByRole("button", { name: "Confirm undo" })); + await screen.findByText("Some files were restored. The remaining files have not been undone."); + expect(screen.getByText("1 path restored")).toBeInTheDocument(); + expect(screen.getByText("1 path remaining")).toBeInTheDocument(); + expect(screen.queryByRole("heading", { name: "This turn’s changes were undone" })).not.toBeInTheDocument(); + await waitFor(() => expect(screen.getByRole("button", { name: "Check again" })).toBeEnabled()); +}); diff --git a/web/src/v2/components/turn-change-undo.tsx b/web/src/v2/components/turn-change-undo.tsx new file mode 100644 index 000000000..07e1a90f8 --- /dev/null +++ b/web/src/v2/components/turn-change-undo.tsx @@ -0,0 +1,106 @@ +import { useId, useRef, useState } from "react"; +import { createPortal } from "react-dom"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { Check, Loader2, Undo2, X } from "lucide-react"; +import type { APIClient } from "../../api/client"; +import type { ThreadChangeSummary } from "../../api/thread-changes"; +import { useModalFocusTrap } from "../../hooks/use-modal-focus-trap"; +import { useLocale } from "../../lib/locale"; +import { v2QueryKeys } from "../query-keys"; + +export function TurnChangeUndo({ client, summary }: { client: APIClient; summary: ThreadChangeSummary }) { + const { t } = useLocale(); + const { thread_id: threadID, run_id: runID, source_event_id: sourceEventID } = summary; + const queryClient = useQueryClient(); + const [open, setOpen] = useState(false); + const trigger = useRef(null); + const cancel = useRef(null); + const titleID = useId(); + const descriptionID = useId(); + const previewKey = ["thread", threadID, "turn-undo", sourceEventID]; + const preview = useQuery({ queryKey: previewKey, + queryFn: ({ signal }) => client.undoThreadChanges(threadID, runID, sourceEventID, false, signal), + enabled: open, retry: false, refetchOnWindowFocus: false }); + const undo = useMutation({ + mutationFn: () => client.undoThreadChanges(threadID, runID, sourceEventID, true), + onSuccess: (result) => { + queryClient.setQueryData(previewKey, result); + queryClient.setQueryData(["thread", threadID, "turn-changes", sourceEventID], + (current) => current ? { ...current, undo: result } : current); + void queryClient.invalidateQueries({ queryKey: ["run", runID] }); + void queryClient.invalidateQueries({ queryKey: v2QueryKeys.thread(threadID) }); + }, + }); + const busy = undo.isPending; + const dialog = useModalFocusTrap(open, () => setOpen(false), busy, cancel, + { isolateBackground: true, returnFocusRef: trigger }); + const result = summary.undo?.state === "completed" ? summary.undo : preview.data ?? summary.undo; + const completed = result?.state === "completed"; + const checking = preview.isFetching; + const error = preview.isError || undo.isError; + const canUndo = result?.state === "available" && !checking && !error && !busy; + + const reasonLabel = (reason: string) => { + switch (reason) { + case "files_changed": return t("文件在本轮之后发生了修改，请先检查当前内容。", "This file changed after this turn. Review its current content first."); + case "run_busy": return t("任务正在执行，结束后才能撤销。", "Wait for the task to stop before undoing changes."); + case "incomplete_history": return t("缺少完整的原始编辑记录，无法恢复。", "The original edit records are incomplete and cannot be restored."); + case "permission_denied": return t("当前没有恢复文件的权限。", "File restore permission is unavailable."); + case "workspace_changed": return t("工作目录已发生变化，请先检查当前目录。", "The working directory has changed. Check the current directory first."); + case "restore_failed": return t("文件恢复未完成，请检查下列文件状态。", "The file restore did not complete. Check the file states below."); + default: return t("当前状态无法安全恢复，请检查文件与任务状态后重试。", "Check the file and task state before retrying the restore."); + } + }; + + return <> + + {open && createPortal(
{ + if (event.target === event.currentTarget && !busy) setOpen(false); + }}> +
+
+

{completed ? t("本轮改动已撤销", "This turn’s changes were undone") : t("撤销本轮改动", "Undo this turn’s changes")}

+ +
+
+

{completed + ? t("已恢复本轮修改前的文件内容。", "The files have been restored to their state before this turn.") + : t(`将恢复这轮对 ${summary.file_count} 个文件的修改。`, `Undo this turn’s changes to ${summary.file_count} ${summary.file_count === 1 ? "file" : "files"}.`)}

+ {(checking || busy) &&

} + {!checking && result?.state === "available" && !busy &&

+ {t("确认前已检查当前文件；如有后续修改，将停止并显示冲突。", "Current files have been checked. Later edits will stop the restore and be shown as conflicts.")}

} + {!checking && result?.state === "partial" &&

{t("部分文件已恢复，其余文件尚未撤销。", "Some files were restored. The remaining files have not been undone.")}

} + {!checking && result?.state === "applying" && !busy &&

{t("上次撤销尚未确认完成，请重新检查结果。", "The previous undo has not completed. Check its result again.")}

} + {error &&

{undo.isError + ? t("暂时无法确认撤销结果。请重新检查后继续。", "The undo result could not be confirmed. Check again before continuing.") + : t("无法检查撤销状态，请重试。", "The undo state could not be checked. Try again.")}

} + {!checking && result && result.conflicts.length > 0 &&
    + {result.conflicts.map((conflict, index) =>
  • + {conflict.path && {conflict.path}}{reasonLabel(conflict.reason)} +
    • )} +
} + {!completed && result && result.restored_paths.length > 0 &&
+ {t(`已恢复 ${result.restored_paths.length} 个路径`, `${result.restored_paths.length} ${result.restored_paths.length === 1 ? "path" : "paths"} restored`)} +
    {result.restored_paths.map((path) =>
  • {path}
    • )}
+
} + {!completed && result && result.restored_paths.length > 0 && result.remaining_paths.length > 0 &&
+ {t(`尚未恢复 ${result.remaining_paths.length} 个路径`, `${result.remaining_paths.length} ${result.remaining_paths.length === 1 ? "path" : "paths"} remaining`)} +
    {result.remaining_paths.map((path) =>
  • {path}
    • )}
+
} +
+
+ {!completed && (canUndo || busy ? + : )} +
+
+
, document.body)} + ; +} diff --git a/web/src/v2/components/turn-changes.css b/web/src/v2/components/turn-changes.css index ae53bf9fe..21ec1ad6f 100644 --- a/web/src/v2/components/turn-changes.css +++ b/web/src/v2/components/turn-changes.css @@ -10,6 +10,10 @@ .v2-change-removed { color: var(--prayu-diff-delete); } .v2-change-unavailable { color: var(--v2-muted); font-size: 12px; white-space: nowrap; } .v2-turn-changes-review { border: 1px solid var(--v2-border); border-radius: 8px; padding: 5px 10px; background: transparent; flex-shrink: 0; } +.v2-turn-changes-actions { display: flex; align-items: center; gap: 8px; flex-shrink: 0; } +.v2-turn-changes-undo { display: inline-flex; align-items: center; justify-content: center; gap: 5px; background: transparent; border: 0; border-radius: 6px; padding: 6px 7px; white-space: nowrap; } +.v2-turn-changes-undo:is(:disabled, [aria-disabled="true"]) { color: var(--v2-muted); cursor: default; } +.v2-turn-changes-undo:is(:disabled, [aria-disabled="true"]):hover { background: transparent; } .v2-turn-changes button:hover, .v2-change-dialog button:hover { background: var(--v2-control-hover); } .v2-assistant-turn .v2-turn-changes-files { list-style: none; padding: 6px 0; margin: 0; border-top: 1px solid var(--v2-border); } .v2-assistant-turn .v2-turn-changes-files li { margin: 0; } @@ -19,7 +23,7 @@ .v2-assistant-turn .v2-turn-changes-scope { margin: 0; padding: 5px 16px 8px; color: var(--v2-muted); font-size: 12px; line-height: 1.5; } .v2-change-load-error { display: flex; align-items: center; gap: 8px; color: var(--v2-muted); font-size: 12px; margin-top: 12px; } .v2-change-load-error button { border: 0; background: transparent; text-decoration: underline; } -.v2-change-overlay { position: fixed; inset: 0; z-index: 100; background: rgb(0 0 0 / 36%); display: grid; place-items: center; padding: 24px; } +.v2-change-overlay { position: fixed; inset: 0; z-index: 500; background: rgb(0 0 0 / 36%); display: grid; place-items: center; padding: 24px; } .v2-change-dialog { width: min(1040px, 100%); height: min(720px, 88dvh); display: flex; flex-direction: column; color: var(--v2-ink); background: var(--prayu-surface-solid); border: 1px solid var(--v2-border-strong); border-radius: 14px; box-shadow: 0 16px 60px rgb(0 0 0 / 24%); font: 13px/1.6 var(--prayu-font-ui); overflow: hidden; } .v2-change-dialog-heading { display: flex; align-items: center; gap: 10px; padding: 14px 18px; border-bottom: 1px solid var(--v2-border); } .v2-change-dialog-heading h2 { font: inherit; font-size: 15px; font-weight: 600; margin: 0; flex: 1; } @@ -42,8 +46,28 @@ .v2-change-diff-line.is-hunk, .v2-change-diff-line.is-meta { color: var(--v2-muted); } .v2-change-more-lines { margin: 12px 16px; padding: 6px 10px; border: 1px solid var(--v2-border); border-radius: 6px; background: var(--v2-control); } .v2-change-dialog > footer { padding: 9px 18px; color: var(--v2-muted); font-size: 12px; border-top: 1px solid var(--v2-border); } +.v2-dialog.v2-turn-undo-dialog { width: min(460px, 100%); min-width: 0; max-width: 100%; max-height: 88dvh; display: flex; flex-direction: column; color: var(--v2-ink); background: var(--prayu-surface-solid); backdrop-filter: none; font: 13px/1.6 var(--prayu-font-ui); } +.v2-turn-undo-dialog button { color: inherit; font: inherit; cursor: pointer; } +.v2-turn-undo-dialog button:disabled { cursor: default; opacity: .6; } +.v2-turn-undo-dialog button:focus-visible { outline: 2px solid var(--v2-accent); outline-offset: 2px; } +.v2-turn-undo-body { padding: 16px 18px; overflow: auto; } +.v2-turn-undo-body p { margin: 0 0 12px; } +.v2-turn-undo-body p:last-child { margin-bottom: 0; } +.v2-turn-undo-note, .v2-turn-undo-status { color: var(--v2-muted); } +.v2-turn-undo-status { display: flex; align-items: center; gap: 7px; } +.v2-turn-undo-conflicts { list-style: none; padding: 0; margin: 14px 0 0; } +.v2-turn-undo-conflicts li { display: flex; flex-direction: column; gap: 3px; padding-block: 10px; border-top: 1px solid var(--v2-border); } +.v2-turn-undo-conflicts span { color: var(--v2-muted); } +.v2-turn-undo-conflicts code, .v2-turn-undo-paths code { font: inherit; overflow-wrap: anywhere; } +.v2-turn-undo-paths { margin-top: 14px; } +.v2-turn-undo-paths summary { cursor: pointer; } +.v2-turn-undo-error { color: var(--v2-danger); } +.v2-turn-undo-spinner { animation: v2-undo-spin 1s linear infinite; } +@keyframes v2-undo-spin { to { transform: rotate(360deg); } } +@media (prefers-reduced-motion: reduce) { .v2-turn-undo-spinner { animation: none; } } @media (max-width: 600px) { - .v2-turn-changes-heading { gap: 9px; padding: 12px; } + .v2-turn-changes-heading { flex-wrap: wrap; gap: 9px; padding: 12px; } + .v2-turn-changes-heading > div { flex-basis: 140px; } .v2-turn-changes-icon { display: none; } .v2-turn-changes-files button { gap: 10px; padding: 8px 12px; } .v2-turn-changes-expand { padding-inline: 12px; } diff --git a/web/src/v2/components/turn-changes.tsx b/web/src/v2/components/turn-changes.tsx index 0370c66de..f0b80338f 100644 --- a/web/src/v2/components/turn-changes.tsx +++ b/web/src/v2/components/turn-changes.tsx @@ -7,6 +7,7 @@ import type { ThreadChangeFile, ThreadChangeSummary } from "../../api/thread-cha import { parseUnifiedDiff } from "../../components/unified-diff"; import { useModalFocusTrap } from "../../hooks/use-modal-focus-trap"; import { useLocale } from "../../lib/locale"; +import { TurnChangeUndo } from "./turn-change-undo"; import "./turn-changes.css"; function ChangeCounts({ value }: { value: Pick }) { @@ -112,7 +113,10 @@ export function V2TurnChanges({ client, threadID, runID, sourceEventID }: {

{summary.files.length ? t(`已编辑 ${summary.file_count} 个文件`, `${summary.file_count} ${summary.file_count === 1 ? "file" : "files"} edited`) : t("本轮改动", "Changes in this turn")}

- {summary.files.length > 0 && } + {summary.files.length > 0 && + {client.hasWorkspaceCheckpointControl && typeof client.undoThreadChanges === "function" && } + + }
    {(expanded ? summary.files : summary.files.slice(0, 3)).map((file, index) =>