diff --git a/internal/app/app.go b/internal/app/app.go index 5473557..dcd477f 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -154,9 +154,21 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { }) wsRepo := workspace.NewPostgresRepository(pool) - // 进程启动时把卡死的 cloning/syncing 复位为 error，避免行永远卡住 - if err := wsRepo.ResetStuckSyncStatuses(ctx); err != nil { + // 进程启动时把卡死的 cloning/syncing 复位为 error，避免行永远卡住； + // 每个被复位的工作区写一条 workspace.sync_aborted_on_restart audit。 + if abortedIDs, err := wsRepo.ResetStuckSyncStatuses(ctx); err != nil { log.Warn("reset stuck workspace sync statuses", "err", err.Error()) + } else { + for _, wsID := range abortedIDs { + if rerr := auditRec.Record(ctx, audit.Entry{ + Action: "workspace.sync_aborted_on_restart", + TargetType: "workspace", + TargetID: wsID.String(), + }); rerr != nil { + log.Warn("workspace.sync_aborted_on_restart audit failed", + "workspace_id", wsID, "err", rerr.Error()) + } + } } pa := projectAccessAdapter{p: projectSvc} @@ -614,7 +626,7 @@ func (a worktreePruneAdapter) enrichWorktrees(ctx context.Context, rows []*works } func (a worktreePruneAdapter) enrichOne(ctx context.Context, w *workspace.Worktree) runners.WorktreeRow { - row := runners.WorktreeRow{ID: w.ID, Branch: w.Branch, Path: w.Path} + row := runners.WorktreeRow{ID: w.ID, WorkspaceID: w.WorkspaceID, Branch: w.Branch, Path: w.Path} ws, err := a.wsRepo.GetWorkspaceByID(ctx, w.WorkspaceID) if err != nil { a.log.Warn("worktree_prune.adapter.lookup_workspace_failed", diff --git a/internal/app/integration_test.go b/internal/app/integration_test.go index 07e2210..638b769 100644 --- a/internal/app/integration_test.go +++ b/internal/app/integration_test.go @@ -345,7 +345,8 @@ func TestPlan3_WorkspaceClosedLoop(t *testing.T) { }) wsRepo := workspace.NewPostgresRepository(pool) - require.NoError(t, wsRepo.ResetStuckSyncStatuses(ctx)) + _, err = wsRepo.ResetStuckSyncStatuses(ctx) + require.NoError(t, err) pa := projectAccessAdapterTest{p: projectSvc} wsSvc := workspace.NewWorkspaceService(wsRepo, auditRec, enc, gitr, pa, nil, dataDir) diff --git a/internal/jobs/runners/worktree_prune.go b/internal/jobs/runners/worktree_prune.go index 8222739..a8f9b30 100644 --- a/internal/jobs/runners/worktree_prune.go +++ b/internal/jobs/runners/worktree_prune.go @@ -15,11 +15,12 @@ import ( // MainPath come from a JOIN performed in the production adapter (workspace // + project), not from the worktrees table directly. type WorktreeRow struct { - ID uuid.UUID - OwnerID uuid.UUID - Branch string - MainPath string - Path string + ID uuid.UUID + WorkspaceID uuid.UUID + OwnerID uuid.UUID + Branch string + MainPath string + Path string } // WorktreePruneRepo is the minimal repository surface used by this runner. @@ -128,7 +129,9 @@ func (r *WorktreePrune) executionPhase(ctx context.Context) { TargetType: "worktree", TargetID: w.ID.String(), Metadata: map[string]any{ - "branch": w.Branch, + "workspace_id": w.WorkspaceID.String(), + "branch": w.Branch, + "path": w.Path, }, }); rerr != nil { r.log.Warn("worktree_prune.audit_failed", "worktree_id", w.ID, "err", rerr.Error()) diff --git a/internal/jobs/runners/worktree_prune_test.go b/internal/jobs/runners/worktree_prune_test.go index 3243c04..576b4a9 100644 --- a/internal/jobs/runners/worktree_prune_test.go +++ b/internal/jobs/runners/worktree_prune_test.go @@ -10,10 +10,20 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/yan1h/agent-coding-workflow/internal/audit" "github.com/yan1h/agent-coding-workflow/internal/infra/notify" "github.com/yan1h/agent-coding-workflow/internal/jobs/runners" ) +type recordingRecorder struct { + entries []audit.Entry +} + +func (r *recordingRecorder) Record(_ context.Context, e audit.Entry) error { + r.entries = append(r.entries, e) + return nil +} + // fakeWorktreeRepo is a fresh fake distinct from fakeWorkspaceFetchRepo. type fakeWorktreeRepo struct { warnList []runners.WorktreeRow @@ -172,6 +182,37 @@ func TestWorktreePrune_NoRowsIsNoOp(t *testing.T) { assert.Empty(t, disp.msgs) } +func TestWorktreePrune_AuditMetadataContainsWorkspaceIDAndPath(t *testing.T) { + t.Parallel() + id := uuid.New() + workspaceID := uuid.New() + ownerID := uuid.New() + row := runners.WorktreeRow{ + ID: id, WorkspaceID: workspaceID, OwnerID: ownerID, + Branch: "feat-audit", MainPath: "/main", Path: "/main/.worktrees/feat-audit", + } + + repo := &fakeWorktreeRepo{ + pruneList: []runners.WorktreeRow{row}, + tryStartResult: row, + } + gitr := &fakePruner{} + rec := &recordingRecorder{} + cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour} + + r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger(), rec) + r.Run(context.Background()) + + require.Len(t, rec.entries, 1) + e := rec.entries[0] + assert.Equal(t, "worktree.pruned", e.Action) + assert.Equal(t, "worktree", e.TargetType) + assert.Equal(t, id.String(), e.TargetID) + assert.Equal(t, workspaceID.String(), e.Metadata["workspace_id"]) + assert.Equal(t, "feat-audit", e.Metadata["branch"]) + assert.Equal(t, "/main/.worktrees/feat-audit", e.Metadata["path"]) +} + func TestWorktreePrune_CASLostSkipsSilently(t *testing.T) { t.Parallel() id := uuid.New() diff --git a/internal/workspace/queries/workspaces.sql b/internal/workspace/queries/workspaces.sql index f06a329..4f8d959 100644 --- a/internal/workspace/queries/workspaces.sql +++ b/internal/workspace/queries/workspaces.sql @@ -35,12 +35,13 @@ RETURNING *; -- name: DeleteWorkspace :exec DELETE FROM workspaces WHERE id = $1; --- name: ResetStuckSyncStatuses :exec +-- name: ResetStuckSyncStatuses :many UPDATE workspaces SET sync_status = 'error', last_sync_error = 'process_restarted_during_' || sync_status, updated_at = now() -WHERE sync_status IN ('cloning', 'syncing'); +WHERE sync_status IN ('cloning', 'syncing') +RETURNING id, last_sync_error; -- name: ListFetchTargets :many SELECT w.id, p.owner_id, w.name, w.main_path diff --git a/internal/workspace/repository.go b/internal/workspace/repository.go index 2936cfc..09bd8c4 100644 --- a/internal/workspace/repository.go +++ b/internal/workspace/repository.go @@ -44,7 +44,9 @@ type Repository interface { UpdateWorkspaceCore(ctx context.Context, id uuid.UUID, name, description, defaultBranch string) (*Workspace, error) UpdateWorkspaceSyncStatus(ctx context.Context, id uuid.UUID, st SyncStatus, syncedAt *time.Time, errMsg string) (*Workspace, error) DeleteWorkspace(ctx context.Context, id uuid.UUID) error - ResetStuckSyncStatuses(ctx context.Context) error + // ResetStuckSyncStatuses 把启动时残留的 cloning/syncing 复位为 error， + // 返回被复位的工作区 ID 列表（按发生顺序），由调用方写 audit。 + ResetStuckSyncStatuses(ctx context.Context) ([]uuid.UUID, error) // Worktree (no-tx 部分) GetWorktreeByID(ctx context.Context, id uuid.UUID) (*Worktree, error) @@ -240,11 +242,16 @@ func (r *pgRepo) DeleteWorkspace(ctx context.Context, id uuid.UUID) error { return nil } -func (r *pgRepo) ResetStuckSyncStatuses(ctx context.Context) error { - if err := r.q.ResetStuckSyncStatuses(ctx); err != nil { - return errs.Wrap(err, errs.CodeInternal, "reset stuck sync") +func (r *pgRepo) ResetStuckSyncStatuses(ctx context.Context) ([]uuid.UUID, error) { + rows, err := r.q.ResetStuckSyncStatuses(ctx) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "reset stuck sync") } - return nil + ids := make([]uuid.UUID, 0, len(rows)) + for _, row := range rows { + ids = append(ids, fromPgUUID(row.ID)) + } + return ids, nil } // ===== Workspace fetch (jobs runner) ===== diff --git a/internal/workspace/sqlc/workspaces.sql.go b/internal/workspace/sqlc/workspaces.sql.go index 2e1cdca..0a587a3 100644 --- a/internal/workspace/sqlc/workspaces.sql.go +++ b/internal/workspace/sqlc/workspaces.sql.go @@ -241,17 +241,38 @@ func (q *Queries) MarkSyncingCAS(ctx context.Context, id pgtype.UUID) (int64, er return result.RowsAffected(), nil } -const resetStuckSyncStatuses = `-- name: ResetStuckSyncStatuses :exec +const resetStuckSyncStatuses = `-- name: ResetStuckSyncStatuses :many UPDATE workspaces SET sync_status = 'error', last_sync_error = 'process_restarted_during_' || sync_status, updated_at = now() WHERE sync_status IN ('cloning', 'syncing') +RETURNING id, last_sync_error ` -func (q *Queries) ResetStuckSyncStatuses(ctx context.Context) error { - _, err := q.db.Exec(ctx, resetStuckSyncStatuses) - return err +type ResetStuckSyncStatusesRow struct { + ID pgtype.UUID `json:"id"` + LastSyncError *string `json:"last_sync_error"` +} + +func (q *Queries) ResetStuckSyncStatuses(ctx context.Context) ([]ResetStuckSyncStatusesRow, error) { + rows, err := q.db.Query(ctx, resetStuckSyncStatuses) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ResetStuckSyncStatusesRow + for rows.Next() { + var i ResetStuckSyncStatusesRow + if err := rows.Scan(&i.ID, &i.LastSyncError); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil } const updateWorkspaceCore = `-- name: UpdateWorkspaceCore :one diff --git a/internal/workspace/workspace_service_test.go b/internal/workspace/workspace_service_test.go index 5b4c46c..a9f8437 100644 --- a/internal/workspace/workspace_service_test.go +++ b/internal/workspace/workspace_service_test.go @@ -75,7 +75,9 @@ func (f *fakeRepo) DeleteWorkspace(_ context.Context, id uuid.UUID) error { delete(f.wss, id) return nil } -func (f *fakeRepo) ResetStuckSyncStatuses(_ context.Context) error { return nil } +func (f *fakeRepo) ResetStuckSyncStatuses(_ context.Context) ([]uuid.UUID, error) { + return nil, nil +} func (f *fakeRepo) ListFetchTargets(_ context.Context) ([]FetchTarget, error) { return nil, nil } func (f *fakeRepo) MarkSyncing(_ context.Context, _ uuid.UUID) (bool, error) { return false, nil } func (f *fakeRepo) MarkFetchResult(_ context.Context, _ uuid.UUID, _ bool, _ string) error {