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,