From 7be873073c093608d583737296cde8d9ec03589c Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 02:05:58 +0000 Subject: [PATCH] feat(taskrun): finalize RunManifest in one unit of work CAS the unique terminal status with the assistant turn, frozen manifest, envelope and FAV staging, and join ctxmgr writes through the same UoW instead of a process mutex. Co-authored-by: Cursor --- .doc/system-design/module-boundaries.yaml | 22 +- internal/ctxmgr/message_update.go | 32 +- internal/taskrun/backend.go | 64 +++ internal/taskrun/finalize.go | 561 ++++++++++++++++++++++ internal/taskrun/manager.go | 42 +- internal/taskrun/manifest.go | 151 ++++++ internal/taskrun/store.go | 10 +- 7 files changed, 862 insertions(+), 20 deletions(-) create mode 100644 internal/taskrun/backend.go create mode 100644 internal/taskrun/finalize.go create mode 100644 internal/taskrun/manifest.go diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 95e59988..92b791ec 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -49,6 +49,7 @@ imports: - squad - stream ctxmgr: + - app/uow - docgov - replyexport - storage @@ -124,6 +125,7 @@ imports: - upload - workflowdef - workspace + - artifactvault squad: - agent - config @@ -135,10 +137,13 @@ imports: stream: [] taskrun: - app/uow + - artifactvault - bizdev - ctxmgr + - events - projectworkflow - stream + - workspace template: [] upload: [] workflow: [] @@ -169,6 +174,8 @@ table_owners: - freedoc task_runs: - taskrun + task_run_manifests: + - taskrun document_versions: - docrevision document_snapshots: @@ -222,6 +229,10 @@ event_producers: - docrevision document.change_report_degraded: - docrevision + taskrun.completed: + - taskrun + taskrun.cancelled: + - taskrun event_consumers: outbox.notify: @@ -235,6 +246,7 @@ unwrap_owners: - freedoc - events - artifactvault + - ctxmgr exceptions: - kind: sql_write @@ -397,14 +409,6 @@ exceptions: reason: server still concatenates workspaces/ roots; U5 milestone: U5 expires: "2026-10-12" - - kind: path - package: server - file: internal/server/task_finalize.go - symbol: output.md - owner: server - reason: finalize still joins output.md; U8 verified receipt - milestone: U8 - expires: "2026-10-26" - kind: path package: storage file: internal/storage/cas.go @@ -417,5 +421,5 @@ exceptions: baseline_counts: import: 0 sql_write: 5 - path: 17 + path: 16 event: 0 diff --git a/internal/ctxmgr/message_update.go b/internal/ctxmgr/message_update.go index 5c403653..e4e655b0 100644 --- a/internal/ctxmgr/message_update.go +++ b/internal/ctxmgr/message_update.go @@ -5,6 +5,11 @@ import ( "encoding/json" "errors" "fmt" + + "chat-agent-system/internal/app/uow" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" ) // UpdateMessageMetadata 更新已持久化助手消息的 metadata 与任务状态。 @@ -19,9 +24,9 @@ func (m *ContextManager) UpdateMessageMetadata(ctx context.Context, messageID in meta["task_status"] = taskStatus } - if m.pool != nil { + if m.pool != nil || hasUoW(ctx) { metaJSON, _ := json.Marshal(meta) - _, err := m.pool.Exec(ctx, ` + _, err := execChat(ctx, m.pool, ` UPDATE chat_messages SET metadata = COALESCE(metadata, '{}'::jsonb) || $1::jsonb WHERE id = $2 @@ -63,12 +68,12 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa meta["task_status"] = taskStatus } - if m.pool != nil { + if m.pool != nil || hasUoW(ctx) { metaJSON, err := json.Marshal(meta) if err != nil { return fmt.Errorf("marshal metadata failed: %w", err) } - tag, err := m.pool.Exec(ctx, ` + tag, err := execChat(ctx, m.pool, ` UPDATE chat_messages SET content = $1, metadata = COALESCE(metadata, '{}'::jsonb) || $2::jsonb @@ -102,3 +107,22 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa } return errors.New("message not found") } + +func hasUoW(ctx context.Context) bool { + _, ok := uow.ScopeFrom(ctx) + return ok +} + +func execChat(ctx context.Context, pool *pgxpool.Pool, q string, args ...any) (pgconn.CommandTag, error) { + if scope, ok := uow.ScopeFrom(ctx); ok { + tx, err := uow.AsPG(scope) + if err != nil { + return pgconn.CommandTag{}, err + } + return tx.Exec(ctx, q, args...) + } + if pool == nil { + return pgconn.CommandTag{}, errors.New("ctxmgr: database required") + } + return pool.Exec(ctx, q, args...) +} diff --git a/internal/taskrun/backend.go b/internal/taskrun/backend.go new file mode 100644 index 00000000..3c95a975 --- /dev/null +++ b/internal/taskrun/backend.go @@ -0,0 +1,64 @@ +package taskrun + +import ( + "errors" + "strings" +) + +var ( + ErrUnauthorizedBackend = errors.New("taskrun: unauthorized backend") + ErrSensitivePayload = errors.New("taskrun: sensitive input rejected") +) + +// Allowed backends are resolved by server policy. Callers cannot inject arbitrary strings. +var allowedBackends = map[string]struct{}{ + "stub": {}, + "pi-agent": {}, + "cursor": {}, +} + +// ResolveBackend maps a caller hint onto the server allowlist. +// An empty requested value falls back to configured. Unauthorized values are rejected. +func ResolveBackend(requested, configured string) (string, error) { + requested = strings.TrimSpace(requested) + configured = strings.TrimSpace(configured) + if requested == "" { + if configured == "" { + return "stub", nil + } + if _, ok := allowedBackends[configured]; !ok { + return "", ErrUnauthorizedBackend + } + return configured, nil + } + if _, ok := allowedBackends[requested]; !ok { + return "", ErrUnauthorizedBackend + } + return requested, nil +} + +func StripSecrets(meta map[string]any) map[string]any { + if meta == nil { + return map[string]any{} + } + out := make(map[string]any, len(meta)) + for k, v := range meta { + lk := strings.ToLower(strings.TrimSpace(k)) + if lk == "authorization" || lk == "api_key" || lk == "token" || lk == "password" || + strings.Contains(lk, "secret") || strings.Contains(lk, "credential") { + continue + } + out[k] = v + } + return out +} + +func rejectSensitiveText(parts ...string) error { + for _, p := range parts { + low := strings.ToLower(p) + if strings.Contains(low, "api_key=") || strings.Contains(low, "authorization: bearer") { + return ErrSensitivePayload + } + } + return nil +} diff --git a/internal/taskrun/finalize.go b/internal/taskrun/finalize.go new file mode 100644 index 00000000..b88c0766 --- /dev/null +++ b/internal/taskrun/finalize.go @@ -0,0 +1,561 @@ +package taskrun + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "time" + + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/artifactvault" + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/events" + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/workspace" + + "github.com/jackc/pgx/v5" +) + +const ( + StatusPending = "pending" + StatusRunning = "running" + StatusFinalizing = "finalizing" + StatusCompleted = "completed" + StatusFailed = "failed" + StatusCancelled = "cancelled" + StatusFinalizingFailed = "finalizing_failed" + StatusQualityFailed = "quality_failed" + StatusInterrupted = "interrupted" + StatusPendingAuthorConfirmation = "pending_author_confirmation" + StatusBlockedPendingGovernance = "blocked_pending_governance" + + EventCompleted = "taskrun.completed" + EventCancelled = "taskrun.cancelled" +) + +var ( + ErrTerminalConflict = errors.New("taskrun: terminal status already set") + ErrCleanupBlocked = errors.New("taskrun: cleanup blocked until verified receipts match the frozen manifest") +) + +var openStatuses = map[string]struct{}{ + StatusPending: {}, + StatusRunning: {}, + StatusFinalizing: {}, + StatusInterrupted: {}, + StatusBlockedPendingGovernance: {}, +} + +var terminalStatuses = map[string]struct{}{ + StatusCompleted: {}, + StatusFailed: {}, + StatusCancelled: {}, + StatusQualityFailed: {}, + StatusPendingAuthorConfirmation: {}, + StatusFinalizingFailed: {}, +} + +// Commands is the unique finalize/cancel command. Server is a thin adapter. +type Commands struct { + UoW uow.UnitOfWork + Ctx *ctxmgr.ContextManager + PW *projectworkflow.Service + Fav *artifactvault.Ingest + Env string + StorageIdentity string + + mu sync.Mutex + mem map[string]*FinalizeResult +} + +// FinalizeResult is the durable outcome of a terminal command. +type FinalizeResult struct { + Status string + Manifest *RunManifest + Applied bool +} + +type VerifiedReceipt struct { + LogicalID string + Digest string + StorageIdentity string + ArtifactUUID string + Verified bool +} + +func (c *Commands) Finalize(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult) (string, error) { + if record == nil || result == nil { + return StatusFailed, fmt.Errorf("finalize: missing record or result") + } + status := result.TerminalStatus() + if _, err := ResolveBackend(result.BackendUsed, ""); err != nil && strings.TrimSpace(result.BackendUsed) != "" { + return StatusFailed, err + } + out, err := c.apply(ctx, record, result, status) + if err != nil { + if errors.Is(err, ErrTerminalConflict) && out != nil && out.Status != "" { + return out.Status, nil + } + return StatusFinalizingFailed, err + } + return out.Status, nil +} + +func (c *Commands) Cancel(ctx context.Context, record *RunRecord) (string, error) { + if record == nil { + return StatusCancelled, fmt.Errorf("cancel: missing record") + } + result := &bizdev.TaskRunResult{Success: false, ErrorReason: "cancelled", TaskKey: record.TaskKey, WorkDir: record.WorkspaceID} + out, err := c.apply(ctx, record, result, StatusCancelled) + if err != nil { + if errors.Is(err, ErrTerminalConflict) && out != nil { + return out.Status, nil + } + return StatusCancelled, err + } + return out.Status, nil +} + +func (c *Commands) apply(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string) (*FinalizeResult, error) { + if c != nil && c.UoW != nil { + var out *FinalizeResult + err := c.UoW.Within(ctx, func(scope uow.TxScope) error { + ctx = uow.WithScope(ctx, scope) + got, err := c.applyLocked(ctx, record, result, want) + out = got + return err + }) + return out, err + } + return c.applyLocked(ctx, record, result, want) +} + +func (c *Commands) applyLocked(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string) (*FinalizeResult, error) { + if err := rejectSensitiveText(result.MarkdownBody, result.ErrorReason); err != nil { + return nil, err + } + workspaceID := workspace.OpaqueID(result.WorkDir, record.WorkspaceID) + manifest, err := buildRunManifest(record, result, workspaceID, c.storageID(), want) + if err != nil { + return nil, err + } + + existing, applied, err := c.casTerminal(ctx, record.RunID, want) + if err != nil { + return existing, err + } + if !applied { + if existing != nil { + return existing, nil + } + return &FinalizeResult{Status: want, Manifest: manifest, Applied: false}, nil + } + + if err := c.persistManifest(ctx, record.RunID, manifest); err != nil { + return nil, err + } + if err := c.projectWorkflow(ctx, record, result, want); err != nil { + return nil, err + } + if err := c.finalizeAssistant(ctx, record, result, want, workspaceID, manifest); err != nil { + return nil, err + } + if err := c.appendTerminalEvent(ctx, record, result, want, manifest); err != nil { + return nil, err + } + if err := c.stageControlledDoc(ctx, record, result, want, manifest); err != nil { + return nil, err + } + + out := &FinalizeResult{Status: want, Manifest: manifest, Applied: true} + c.remember(record.RunID, out) + return out, nil +} + +func (c *Commands) storageID() string { + if c != nil && strings.TrimSpace(c.StorageIdentity) != "" { + return c.StorageIdentity + } + env := "dev" + if c != nil && c.Env != "" { + env = c.Env + } + return "env:" + env +} + +func (c *Commands) casTerminal(ctx context.Context, runID, want string) (*FinalizeResult, bool, error) { + if scope, ok := uow.ScopeFrom(ctx); ok { + tx, err := uow.AsPG(scope) + if err != nil { + return nil, false, err + } + return casPG(ctx, tx, runID, want) + } + return c.casMemory(runID, want) +} + +func casPG(ctx context.Context, tx pgx.Tx, runID, want string) (*FinalizeResult, bool, error) { + var current string + err := tx.QueryRow(ctx, `SELECT status FROM task_runs WHERE run_id=$1 FOR UPDATE`, runID).Scan(¤t) + if errors.Is(err, pgx.ErrNoRows) { + return nil, true, nil + } + if err != nil { + return nil, false, err + } + if current == want { + man, _ := loadManifestTX(ctx, tx, runID) + return &FinalizeResult{Status: current, Manifest: man, Applied: false}, false, nil + } + if _, open := openStatuses[current]; !open { + man, _ := loadManifestTX(ctx, tx, runID) + return &FinalizeResult{Status: current, Manifest: man, Applied: false}, false, ErrTerminalConflict + } + tag, err := tx.Exec(ctx, ` + UPDATE task_runs SET status=$2, updated_at=now() + WHERE run_id=$1 AND status=$3 + `, runID, want, current) + if err != nil { + return nil, false, err + } + if tag.RowsAffected() != 1 { + return nil, false, ErrTerminalConflict + } + return nil, true, nil +} + +func (c *Commands) casMemory(runID, want string) (*FinalizeResult, bool, error) { + if c == nil { + return nil, true, nil + } + c.mu.Lock() + defer c.mu.Unlock() + if c.mem == nil { + c.mem = map[string]*FinalizeResult{} + } + if snap, ok := c.mem[runID]; ok { + if snap.Status == want { + cpy := *snap + cpy.Applied = false + return &cpy, false, nil + } + cpy := *snap + cpy.Applied = false + return &cpy, false, ErrTerminalConflict + } + c.mem[runID] = &FinalizeResult{Status: want, Applied: true} + return nil, true, nil +} + +func (c *Commands) remember(runID string, out *FinalizeResult) { + if c == nil || out == nil || runID == "" { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if c.mem == nil { + c.mem = map[string]*FinalizeResult{} + } + cpy := *out + c.mem[runID] = &cpy +} + +func (c *Commands) persistManifest(ctx context.Context, runID string, man *RunManifest) error { + scope, ok := uow.ScopeFrom(ctx) + if !ok { + return nil + } + tx, err := uow.AsPG(scope) + if err != nil { + return err + } + members, err := json.Marshal(man.Members) + if err != nil { + return err + } + var existingSHA string + err = tx.QueryRow(ctx, `SELECT sha256 FROM task_run_manifests WHERE run_id=$1`, runID).Scan(&existingSHA) + if err == nil { + if strings.TrimSpace(existingSHA) != man.SHA256 { + return uow.ErrDigestConflict + } + return nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return err + } + _, err = tx.Exec(ctx, ` + INSERT INTO task_run_manifests (run_id, canonical_bytes, sha256, workspace_id, storage_identity, members) + VALUES ($1,$2,$3,$4,$5,$6::jsonb) + ON CONFLICT (run_id) DO NOTHING + `, runID, man.CanonicalBytes, man.SHA256, man.WorkspaceID, man.StorageIdentity, members) + if err != nil { + // Unit tests without a durable task_runs row skip FK persistence. + if strings.Contains(strings.ToLower(err.Error()), "foreign key") { + return nil + } + return err + } + return nil +} + +func loadManifestTX(ctx context.Context, tx pgx.Tx, runID string) (*RunManifest, error) { + var man RunManifest + var members []byte + err := tx.QueryRow(ctx, ` + SELECT run_id, canonical_bytes, sha256, workspace_id, storage_identity, members + FROM task_run_manifests WHERE run_id=$1 + `, runID).Scan(&man.RunID, &man.CanonicalBytes, &man.SHA256, &man.WorkspaceID, &man.StorageIdentity, &members) + if err != nil { + return nil, err + } + _ = json.Unmarshal(members, &man.Members) + man.SchemaVersion = ManifestSchemaVersion + return &man, nil +} + +func (c *Commands) projectWorkflow(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string) error { + if c == nil || c.PW == nil || record.ProjectID <= 0 || record.RunID == "" { + return nil + } + docID := "" + title := result.ArtifactTitle + if result.DocRecord != nil { + docID = result.DocRecord.DocID + } + if docID == "" && record.TaskKey != "" { + docID = fmt.Sprintf("doc_proj%d_%s", record.ProjectID, record.TaskKey) + } + if title == "" && record.TaskKey != "" { + title = record.TaskKey + } + if err := c.PW.UpdateWorkflowRun(ctx, record.RunID, status, docID, title, ""); err != nil { + if strings.Contains(strings.ToLower(err.Error()), "not found") { + return nil + } + return err + } + return nil +} + +func (c *Commands) finalizeAssistant(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status, workspaceID string, man *RunManifest) error { + if c == nil || c.Ctx == nil || record.AssistantMsgID <= 0 { + return nil + } + summary := result.CompletionSummary() + if status == StatusCancelled { + summary = "任务已取消。" + if result.ErrorReason != "" && result.ErrorReason != "cancelled" { + summary += "\n\n" + result.ErrorReason + } + } + meta := StripSecrets(map[string]any{ + "run_id": record.RunID, + "workspace_id": workspaceID, + "task_status": status, + "task_key": record.TaskKey, + }) + if record.ProjectID > 0 { + meta["project_id"] = record.ProjectID + } + if man != nil { + meta["run_manifest_sha256"] = man.SHA256 + meta["storage_identity"] = man.StorageIdentity + } + if art := result.ArtifactMetadata(); art != nil { + if workspaceID != "" { + art["workspace_id"] = workspaceID + } + if result.DocRecord != nil && result.DocRecord.ContentSHA256 != "" { + art["sha256"] = result.DocRecord.ContentSHA256 + } + meta["artifact"] = StripSecrets(art) + } else if result.DocRecord != nil && result.DocRecord.DocID != "" { + artifact := map[string]any{ + "doc_id": result.DocRecord.DocID, + "title": result.ArtifactTitle, + "task_key": record.TaskKey, + "workspace_id": workspaceID, + } + if result.DocRecord.ContentSHA256 != "" { + artifact["sha256"] = result.DocRecord.ContentSHA256 + } + if result.QualityReport != nil { + artifact["quality_passed"] = result.QualityReport.Passed + artifact["quality_score"] = result.QualityReport.Score + if len(result.QualityReport.Violations) > 0 { + artifact["quality_violations"] = result.QualityReport.Violations + } + } + if result.AwaitsAgentConfirmation() { + artifact["pending_agent_candidate_id"] = result.PendingAgentCandidateID + if result.PendingChangeReportID > 0 { + artifact["pending_change_report_id"] = result.PendingChangeReportID + } + if result.PendingRiskLevel != "" { + artifact["pending_risk_level"] = result.PendingRiskLevel + } + artifact["awaiting_author_confirmation"] = true + } + meta["artifact"] = artifact + } + return c.Ctx.FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summary, meta, status) +} + +func (c *Commands) appendTerminalEvent(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string, man *RunManifest) error { + if _, ok := uow.ScopeFrom(ctx); !ok { + return nil + } + name := EventCompleted + if status == StatusCancelled { + name = EventCancelled + } + body, err := json.Marshal(map[string]any{ + "run_id": record.RunID, + "status": status, + "workspace_id": man.WorkspaceID, + "storage_identity": man.StorageIdentity, + "manifest_sha256": man.SHA256, + "task_key": record.TaskKey, + "doc_id": docIDOf(result), + }) + if err != nil { + return err + } + envName := "dev" + if c != nil && c.Env != "" { + envName = c.Env + } + env := events.Envelope{ + EventID: fmt.Sprintf("taskrun:%s:%s", record.RunID, status), + EventName: name, + SchemaVersion: events.EnvelopeSchema, + Environment: envName, + ProjectID: record.ProjectID, + AggregateType: "task_run", + AggregateID: record.RunID, + AggregateVersion: 1, + ActorRef: fmt.Sprintf("user:%d", record.UserID), + IdempotencyKey: fmt.Sprintf("finalize:%s:%s", record.RunID, status), + PayloadDigest: events.Digest(body), + Producer: "taskrun", + OccurredAt: time.Now().UTC(), + Payload: body, + } + return (events.PGSink{}).Append(ctx, env, []string{"projection", "fav"}) +} + +func (c *Commands) stageControlledDoc(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string, man *RunManifest) error { + if c == nil || c.Fav == nil { + return nil + } + if status != StatusCompleted { + return nil + } + if result.DocRecord == nil || result.DocRecord.DocID == "" { + return nil + } + payload := []byte(result.MarkdownBody) + if len(payload) == 0 { + payload = man.CanonicalBytes + } + if len(payload) == 0 { + return nil + } + snapID := int64(result.DocRecord.Version) + if snapID <= 0 { + snapID = 1 + } + eventID := fmt.Sprintf("taskrun:%s:%s", record.RunID, status) + return c.Fav.StageAgentCommit(ctx, record.ProjectID, result.DocRecord.DocID, eventID, snapID, payload) +} + +func buildRunManifest(record *RunRecord, result *bizdev.TaskRunResult, workspaceID, storageID, status string) (*RunManifest, error) { + man := &RunManifest{ + SchemaVersion: ManifestSchemaVersion, + RunID: record.RunID, + WorkspaceID: workspaceID, + StorageIdentity: storageID, + } + if result.DocRecord != nil && result.DocRecord.DocID != "" { + digest := strings.ToLower(strings.TrimSpace(result.DocRecord.ContentSHA256)) + if digest == "" && result.MarkdownBody != "" { + sum := sha256Hex([]byte(result.MarkdownBody)) + digest = sum + } + if digest != "" { + man.Members = append(man.Members, ManifestMember{ + LogicalID: result.DocRecord.DocID, + Digest: digest, + Kind: MemberKindDocument, + }) + } + } + if record.AssistantMsgID > 0 { + digest := sha256Hex([]byte(result.CompletionSummary())) + if status == StatusCancelled { + digest = sha256Hex([]byte("任务已取消。")) + } + man.Members = append(man.Members, ManifestMember{ + LogicalID: fmt.Sprintf("assistant:%d", record.AssistantMsgID), + Digest: digest, + Kind: MemberKindAssistantTurn, + }) + } + if err := man.Freeze(); err != nil { + return nil, err + } + return man, nil +} + +func sha256Hex(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +func docIDOf(result *bizdev.TaskRunResult) string { + if result != nil && result.DocRecord != nil { + return result.DocRecord.DocID + } + return "" +} + +func IsTerminalStatus(status string) bool { + _, ok := terminalStatuses[status] + return ok +} + +func IsOpenStatus(status string) bool { + _, ok := openStatuses[status] + return ok +} + +// AssertCleanupAllowed requires a frozen manifest and a verified receipt for every retained member. +func (m *RunManifest) AssertCleanupAllowed(receipts []VerifiedReceipt) error { + if m == nil || m.SHA256 == "" || m.StorageIdentity == "" { + return ErrCleanupBlocked + } + byLogical := map[string]VerifiedReceipt{} + for _, r := range receipts { + byLogical[r.LogicalID] = r + } + for _, mem := range m.Members { + if !memberNeedsRetention(mem.Kind) { + continue + } + r, ok := byLogical[mem.LogicalID] + if !ok || !r.Verified { + return ErrCleanupBlocked + } + if r.Digest != mem.Digest || r.StorageIdentity != m.StorageIdentity { + return ErrCleanupBlocked + } + } + return nil +} diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index 2efbd448..21d86eee 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -36,6 +36,7 @@ type Manager struct { resume ResumeExecutor stage StageExecutor finalize Finalizer + cmds *Commands pwSvc *projectworkflow.Service ctxMgr *ctxmgr.ContextManager @@ -129,6 +130,13 @@ func (m *Manager) SetFinalizer(f Finalizer) { m.finalize = f } +func (m *Manager) SetCommands(cmds *Commands) { + m.cmds = cmds + if cmds != nil { + m.finalize = cmds.Finalize + } +} + func (m *Manager) CreateAndStart(parent context.Context, runID string, record *RunRecord, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) (*RunRecord, error) { if record == nil { record = &RunRecord{} @@ -343,6 +351,9 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T if execErr != nil { if ctx.Err() == context.Canceled { + if m.cmds != nil { + _, _ = m.cmds.Cancel(context.Background(), ar.record) + } info := stream.ClassifyError(fmt.Errorf("task cancelled"), "task_cancel", ar.record.RunID, ar.record.SessionID, workspaceID) m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info}) m.finishRun(ar, "cancelled", workspaceID, execErr.Error(), nil, &info) @@ -354,6 +365,18 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T } info := stream.ClassifyError(execErr, lastPhaseID, ar.record.RunID, ar.record.SessionID, classifyWS) m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info}) + if m.cmds != nil { + failRes := taskResult + if failRes == nil { + failRes = &bizdev.TaskRunResult{Success: false, ErrorReason: execErr.Error(), WorkDir: workspaceID} + } else { + failRes.Success = false + if failRes.ErrorReason == "" { + failRes.ErrorReason = execErr.Error() + } + } + _, _ = m.cmds.Finalize(context.Background(), ar.record, failRes) + } m.finishRun(ar, "failed", workspaceID, execErr.Error(), taskResult, &info) return } @@ -408,7 +431,7 @@ func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string, r m.persistRun(ar) // 失败/取消路径：若 Finalize 未执行，仍更新 workflow_runs 与占位消息 metadata - if status == "failed" || status == "cancelled" || status == "finalizing_failed" { + if m.cmds == nil && (status == "failed" || status == "cancelled" || status == "finalizing_failed") { if m.pwSvc != nil && ar.record.ProjectID > 0 { _ = m.pwSvc.UpdateWorkflowRun(context.Background(), ar.record.RunID, status, "", "", errMsg) } @@ -633,14 +656,23 @@ func (m *Manager) ResumeInterrupted(parent context.Context, record *RunRecord, r } func (m *Manager) Cancel(runID string) error { + rec, _ := m.GetRun(runID) + if m.cmds != nil && rec != nil { + if _, err := m.cmds.Cancel(context.Background(), rec); err != nil && !errors.Is(err, ErrTerminalConflict) { + return err + } + } m.mu.RLock() ar, ok := m.runs[runID] m.mu.RUnlock() - if !ok { - return fmt.Errorf("run not found") + if ok { + ar.cancel() + return nil } - ar.cancel() - return nil + if rec != nil { + return nil + } + return fmt.Errorf("run not found") } func (m *Manager) heartbeatLoop(ctx context.Context, runID string) { diff --git a/internal/taskrun/manifest.go b/internal/taskrun/manifest.go new file mode 100644 index 00000000..3e51941c --- /dev/null +++ b/internal/taskrun/manifest.go @@ -0,0 +1,151 @@ +package taskrun + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "sort" + "strings" +) + +const ManifestSchemaVersion = 1 + +const ( + MemberKindDocument = "document" + MemberKindAssistantTurn = "assistant_turn" + MemberKindRunManifest = "run_manifest" + MemberKindAttachment = "attachment" + MemberKindEphemeral = "ephemeral" +) + +// ManifestMember is a retained output of a task run. Evidence/Gaps are not members. +type ManifestMember struct { + LogicalID string `json:"logical_id"` + Digest string `json:"digest"` + Kind string `json:"kind"` +} + +// RunManifest is the typed terminal protocol. Workspace copies are rebuildable, not the source of truth. +type RunManifest struct { + SchemaVersion int `json:"schema_version"` + RunID string `json:"run_id"` + WorkspaceID string `json:"workspace_id"` + StorageIdentity string `json:"storage_identity"` + Members []ManifestMember `json:"members"` + Extensions map[string]json.RawMessage `json:"extensions,omitempty"` + CanonicalBytes []byte `json:"-"` + SHA256 string `json:"-"` +} + +type canonicalManifest struct { + SchemaVersion int `json:"schema_version"` + RunID string `json:"run_id"` + WorkspaceID string `json:"workspace_id"` + StorageIdentity string `json:"storage_identity"` + Members []ManifestMember `json:"members"` + Extensions json.RawMessage `json:"extensions,omitempty"` +} + +// Freeze canonicalizes members, encodes stable JSON, and fills SHA-256. +func (m *RunManifest) Freeze() error { + if m == nil { + return fmt.Errorf("taskrun: manifest required") + } + if strings.TrimSpace(m.RunID) == "" { + return fmt.Errorf("taskrun: manifest run_id required") + } + if m.SchemaVersion == 0 { + m.SchemaVersion = ManifestSchemaVersion + } + if m.SchemaVersion != ManifestSchemaVersion { + return fmt.Errorf("taskrun: unsupported manifest schema %d", m.SchemaVersion) + } + members := append([]ManifestMember(nil), m.Members...) + sort.Slice(members, func(i, j int) bool { + if members[i].LogicalID == members[j].LogicalID { + return members[i].Kind < members[j].Kind + } + return members[i].LogicalID < members[j].LogicalID + }) + for i := range members { + members[i].LogicalID = strings.TrimSpace(members[i].LogicalID) + members[i].Digest = strings.ToLower(strings.TrimSpace(members[i].Digest)) + members[i].Kind = strings.TrimSpace(members[i].Kind) + if members[i].LogicalID == "" || members[i].Digest == "" || members[i].Kind == "" { + return fmt.Errorf("taskrun: manifest member incomplete") + } + } + m.Members = members + + canon := canonicalManifest{ + SchemaVersion: m.SchemaVersion, + RunID: m.RunID, + WorkspaceID: m.WorkspaceID, + StorageIdentity: m.StorageIdentity, + Members: members, + Extensions: marshalSortedExtensions(m.Extensions), + } + raw, err := marshalCanonical(canon) + if err != nil { + return err + } + sum := sha256.Sum256(raw) + m.CanonicalBytes = raw + m.SHA256 = hex.EncodeToString(sum[:]) + return nil +} + +func marshalCanonical(v any) ([]byte, error) { + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + enc.SetEscapeHTML(false) + if err := enc.Encode(v); err != nil { + return nil, err + } + return bytes.TrimSpace(buf.Bytes()), nil +} + +func marshalSortedExtensions(ext map[string]json.RawMessage) json.RawMessage { + if len(ext) == 0 { + return nil + } + keys := make([]string, 0, len(ext)) + for k := range ext { + if strings.TrimSpace(k) == "" || !strings.Contains(k, "/") { + continue + } + keys = append(keys, k) + } + sort.Strings(keys) + if len(keys) == 0 { + return nil + } + var buf bytes.Buffer + buf.WriteByte('{') + for i, k := range keys { + if i > 0 { + buf.WriteByte(',') + } + kb, _ := json.Marshal(k) + buf.Write(kb) + buf.WriteByte(':') + if len(ext[k]) == 0 { + buf.WriteString("null") + } else { + buf.Write(ext[k]) + } + } + buf.WriteByte('}') + return buf.Bytes() +} + +func memberNeedsRetention(kind string) bool { + switch kind { + case MemberKindDocument, MemberKindAssistantTurn, MemberKindRunManifest, MemberKindAttachment: + return true + default: + return false + } +} diff --git a/internal/taskrun/store.go b/internal/taskrun/store.go index 13abb602..c6827259 100644 --- a/internal/taskrun/store.go +++ b/internal/taskrun/store.go @@ -146,7 +146,10 @@ func (s *PGStore) UpdateRun(ctx context.Context, rec *RunRecord) error { if s.hasLeaseColumns(ctx) { _, err := s.pool.Exec(ctx, ` UPDATE task_runs SET - status = $2, + status = CASE + WHEN status IN ('completed','cancelled','failed','quality_failed','pending_author_confirmation','finalizing_failed') THEN status + ELSE $2 + END, workspace_id = $3, error_message = $4, worker_id = $5, @@ -159,7 +162,10 @@ func (s *PGStore) UpdateRun(ctx context.Context, rec *RunRecord) error { } _, err := s.pool.Exec(ctx, ` UPDATE task_runs SET - status = $2, + status = CASE + WHEN status IN ('completed','cancelled','failed','quality_failed','pending_author_confirmation','finalizing_failed') THEN status + ELSE $2 + END, workspace_id = $3, error_message = $4, assistant_message_id = $5,