From 2234af476a552b9207c5ddf6839484bbfa89cf01 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 01:04:32 +0000 Subject: [PATCH 1/7] docs(ops): claim M3 U8 TaskRun RunManifest mutex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Register feature/taskrun-runmanifest on the taskrun mutex so U8 can start without occupying chat/intent or writing U9–U12. Co-authored-by: Cursor --- .doc/ops/CURRENT.md | 7 ++- .../2026-09-15-taskrun-runmanifest-u8.md | 60 +++++++++++++++++++ 2 files changed, 64 insertions(+), 3 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-15-taskrun-runmanifest-u8.md diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 6611776c..cc24229d 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -1,11 +1,11 @@ # 当前焦点 -**更新**: 2026-09-14 +**更新**: 2026-09-15 **production_ready**: no（`CHAT-INTENT-STATE` 等无关 blocker 仍开放。文档账本运行时 UI Round20 **PASS**，不把 fallback change report 写成完整 PI，也不把本轮写成全仓可发布） ## 焦点 -1. **DOC-RELIABILITY**: [M2 已签，继续 M3](../test-reports/2026-09-14-document-reliability-m2-test-report.md)（`refactor/document-reliability-pilot`；M2 最后一句 **继续 M3**；未开工 U8–U12） +1. **DOC-RELIABILITY**: [M3 U8 TaskRun/RunManifest](short-term/2026-09-15-taskrun-runmanifest-u8.md)（`feature/taskrun-runmanifest`；权威停走门 [M2 继续 M3](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；**未签字前禁止 U9–U12**） 2. **CHAT-INTENT**: [对话意图运行时状态](../audit/2026-09-13-conversation-intent-runtime-state.md)（`fix/conversation-intent-state`；与文档账本运行时无关） 3. **CHAT**: [对话历史刷新](../audit/2026-09-13-chat-history-refresh-disappears.md)（认领仍 `in_progress`；gap 行已 PASS，以代码/handoff 为准收敛） @@ -13,6 +13,7 @@ | owner | task | branch | worktree | mutex | status | |-------|------|--------|----------|-------|--------| +| Cursor | taskrun-runmanifest-u8 | feature/taskrun-runmanifest | - | taskrun | in_progress | | Cursor | document-reliability-m1-live-and-m2 | refactor/document-reliability-pilot | .worktrees/refactor-document-reliability-pilot | workspace | done | | Codex | module-decoupling-remediation-plan | feature/module-decoupling-remediation | .worktrees/feature/module-decoupling-remediation | architecture-docs | done | | verifier_antigravity | controlled-content-ui-acceptance | fix/controlled-content-ui-acceptance | .worktrees/fix-controlled-content-ui-acceptance | document-content-ui-acceptance | done | @@ -54,7 +55,7 @@ > 新任务开始前登记。`branch` 必须为 `feature/` 或 `fix/`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。 > workflow-plan 409 与治理鉴权 G1–G5 已收口（见 [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)）。禁止终止或改写项目 143。 -> `document-reliability-pilot` M2 已签（继续 M3）。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未签字前禁止 U8–U12。 +> `document-reliability-pilot` M2 已签（继续 M3）。U8 认领 `mutex=taskrun`。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未签字前禁止 U9–U12。 ## 集成槽位 diff --git a/.doc/ops/short-term/2026-09-15-taskrun-runmanifest-u8.md b/.doc/ops/short-term/2026-09-15-taskrun-runmanifest-u8.md new file mode 100644 index 00000000..a623aaae --- /dev/null +++ b/.doc/ops/short-term/2026-09-15-taskrun-runmanifest-u8.md @@ -0,0 +1,60 @@ +--- +title: "M3 U8 TaskRun 终态与 RunManifest" +status: in_progress +created: 2026-09-15 +updated: 2026-09-15 +owner: Cursor +branch: feature/taskrun-runmanifest +worktree: "-" +mutex: taskrun +--- + +# M3 U8 TaskRun 终态与 RunManifest + +## 目标 + +TaskRun 终态在同一 UoW 内 CAS 写入唯一合法状态，并持久化类型化 RunManifest；重复 finalize / cancel race / SSE 断线不得产生第二条助手消息或第二份 artifact；workspace 清理仅在 verified receipt 之后。 + +## 范围 + +- Goose 00037 `task_run_manifests` + `finalizing` 状态 +- `internal/taskrun` RunManifest 规范字节 / SHA / members +- `taskrun.Commands.Finalize` 唯一 finalize 命令；server 变薄 +- `ctxmgr.FinalizeAssistantTaskMessage` 仅此写路径加入 UoW +- workspace 清理门禁；`ReadArtifact` 替代 `Join(workDir, "output.md")` +- BackendRef 服务端 policy；隔离库并发测试；占槽 rebuild；U8 报告 + +## 非目标 + +- U9 backfill / corpus / 八轨 / 销毁 +- M4–M6；整支 merge FAV；启用 `FAV_DEGRADED_READ` +- 终止或改写项目 143 +- 动 `chat` / `intent` 刷新任务 +- worktree 内 `env-service` + +## 步骤 + +1. [x] CURRENT 认领 `mutex=taskrun`；从本地 `dev` 开 `feature/taskrun-runmanifest` +2. [ ] Goose 00037 Up+Down；`module-boundaries.yaml` 表属主 +3. [ ] Finalize 同 UoW：CAS + 助手 turn + manifest + envelope/staging +4. [ ] 清理门禁与 server 去路径、BackendRef +5. [ ] 隔离库双 pool / race；占槽 rebuild；报告最后一句只能「继续 U9」或「停止扩面」 + +## 验收标准 + +- [ ] 重复 finalize：同一 status、同一 manifest SHA、助手消息 1 条、envelope/staging 不加倍 +- [ ] finalize ∥ cancel：恰好一个合法终态 +- [ ] SSE `done` 不出现在 TX 失败路径；commit 前断线重试不第二 turn +- [ ] 无 verified receipt 则 cleanup FAIL；有 receipt 则可清 workspace 且 staging 仍能取出原 SHA +- [ ] 报告最后一句为「继续 U9」或「停止扩面」；U9 清单项保持未勾 + +## 相关链接 + +- Checklist: [2026-09-14-module-decoupling-refactor-acceptance.md](../checklists/2026-09-14-module-decoupling-refactor-acceptance.md) +- Remediation: [2026-09-14-module-decoupling-refactor-plan.md](../../remediation/2026-09-14-module-decoupling-refactor-plan.md) +- 权威停走门: [2026-09-14-document-reliability-m2-test-report.md](../../test-reports/2026-09-14-document-reliability-m2-test-report.md)（继续 M3） +- Test report: （本轮结束后回链） + +## 证据 / 备注 + +未签字前禁止 U9–U12。合入 + 容器测占集成槽，主仓 `./scripts/env-service.sh dev rebuild`。 From eb75f855a566aec599716b51f515c624787a88fe Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 02:05:33 +0000 Subject: [PATCH 2/7] feat(db): add goose 00037 task_run_manifests Persist typed RunManifest bytes and SHA per run, and allow a finalizing status so U8 can freeze a unique terminal without sharing env databases. Co-authored-by: Cursor --- .../00037_2026-09-15-task-run-manifests.sql | 54 +++++++++++++++++++ internal/db/migrate_test.go | 5 +- internal/db/testdb.go | 10 ++++ 3 files changed, 67 insertions(+), 2 deletions(-) create mode 100644 deploy/postgres/goose/00037_2026-09-15-task-run-manifests.sql diff --git a/deploy/postgres/goose/00037_2026-09-15-task-run-manifests.sql b/deploy/postgres/goose/00037_2026-09-15-task-run-manifests.sql new file mode 100644 index 00000000..c2195c9e --- /dev/null +++ b/deploy/postgres/goose/00037_2026-09-15-task-run-manifests.sql @@ -0,0 +1,54 @@ +-- +goose Up +-- U8: typed RunManifest persistence and optional finalizing status. + +ALTER TABLE task_runs DROP CONSTRAINT IF EXISTS chk_task_run_status; + +ALTER TABLE task_runs ADD CONSTRAINT chk_task_run_status CHECK ( + status IN ( + 'pending', + 'running', + 'finalizing', + 'completed', + 'failed', + 'cancelled', + 'finalizing_failed', + 'quality_failed', + 'interrupted', + 'pending_author_confirmation', + 'blocked_pending_governance' + ) +); + +CREATE TABLE IF NOT EXISTS task_run_manifests ( + run_id TEXT PRIMARY KEY REFERENCES task_runs(run_id) ON DELETE CASCADE, + canonical_bytes BYTEA NOT NULL, + sha256 CHAR(64) NOT NULL, + workspace_id TEXT NOT NULL DEFAULT '', + storage_identity TEXT NOT NULL DEFAULT '', + members JSONB NOT NULL DEFAULT '[]'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uq_task_run_manifests_run UNIQUE (run_id) +); + +CREATE INDEX IF NOT EXISTS idx_task_run_manifests_sha ON task_run_manifests (sha256); + +-- +goose Down +DROP INDEX IF EXISTS idx_task_run_manifests_sha; +DROP TABLE IF EXISTS task_run_manifests; + +ALTER TABLE task_runs DROP CONSTRAINT IF EXISTS chk_task_run_status; + +ALTER TABLE task_runs ADD CONSTRAINT chk_task_run_status CHECK ( + status IN ( + 'pending', + 'running', + 'completed', + 'failed', + 'cancelled', + 'finalizing_failed', + 'quality_failed', + 'interrupted', + 'pending_author_confirmation', + 'blocked_pending_governance' + ) +); diff --git a/internal/db/migrate_test.go b/internal/db/migrate_test.go index 4c3da960..dfd3f360 100644 --- a/internal/db/migrate_test.go +++ b/internal/db/migrate_test.go @@ -100,6 +100,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { "00034_2026-09-14-workspace-lease-fencing.sql", "00035_2026-09-14-durable-event-delivery.sql", "00036_2026-09-14-final-artifact-vault.sql", + "00037_2026-09-15-task-run-manifests.sql", } joined := strings.Join(files, "\n") for _, name := range want { @@ -135,7 +136,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { t.Fatalf("up: %v", err) } ver, err := CurrentVersion(ctx, pool) - if err != nil || ver < 36 { + if err != nil || ver < 37 { t.Fatalf("version after up: %d %v", ver, err) } @@ -148,7 +149,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { if err := goose.DownToContext(ctx, sqlDB, migDir, 33); err != nil { t.Fatalf("down to 33: %v", err) } - for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts"} { + for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts", "task_run_manifests"} { var n int if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM information_schema.tables WHERE table_name=$1`, table).Scan(&n); err != nil { t.Fatal(err) diff --git a/internal/db/testdb.go b/internal/db/testdb.go index 4abbb032..e0a8d35a 100644 --- a/internal/db/testdb.go +++ b/internal/db/testdb.go @@ -49,3 +49,13 @@ func SkipIfNoTestDatabaseURL(t *testing.T) string { } return dsn } + +// RefuseSharedEnvDSN fails if the DSN targets a shared env database. +func RefuseSharedEnvDSN(t *testing.T, dsn string) { + t.Helper() + if strings.Contains(dsn, "5433") || strings.Contains(dsn, "agent_system_dev") || + (strings.Contains(dsn, "5434") && strings.Contains(dsn, "agent_system_test")) || + strings.Contains(dsn, "5435") { + t.Fatal("TEST_DATABASE_URL points at a shared env port; refuse agent_system_dev/test/main") + } +} From 7be873073c093608d583737296cde8d9ec03589c Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 02:05:58 +0000 Subject: [PATCH 3/7] 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, From 2ddaf769ca8f9f9a57c3ed32079d9777833b474a Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 02:06:10 +0000 Subject: [PATCH 4/7] feat(workspace): gate cleanup on verified receipts Hash deliverables through Capability.ReadArtifact, resolve backends on the server, and refuse workspace removal until frozen members match FAV receipts. Co-authored-by: Cursor --- internal/server/handlers.go | 4 +- internal/server/server.go | 18 ++++ internal/server/sse.go | 2 +- internal/server/task_finalize.go | 150 ++++++++++----------------- internal/server/task_run_handlers.go | 17 ++- internal/workspace/cleanup.go | 111 ++++++++++++++++++++ internal/workspace/cleanup_test.go | 63 +++++++++++ internal/workspace/keys.go | 4 + 8 files changed, 266 insertions(+), 103 deletions(-) create mode 100644 internal/workspace/cleanup.go create mode 100644 internal/workspace/cleanup_test.go create mode 100644 internal/workspace/keys.go diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 37eccceb..7f9873e8 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -259,7 +259,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { IntendedUse: req.IntendedUse, TargetMarket: req.TargetMarket, Fields: req.Fields, - Backend: req.Backend, + Backend: resolveTaskBackend(s, req.Backend), AutoRepair: req.AutoRepair, UserID: req.UserID, ProjectID: req.ProjectID, @@ -308,7 +308,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { TargetMarket: req.TargetMarket, Fields: req.Fields, Materials: req.Materials, - Backend: req.Backend, + Backend: resolveTaskBackend(s, req.Backend), AutoRepair: req.AutoRepair, UserID: req.UserID, ProjectID: req.ProjectID, diff --git a/internal/server/server.go b/internal/server/server.go index e9731e9f..f0e703d4 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -11,6 +11,8 @@ import ( "time" "chat-agent-system/internal/agent" + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/artifactvault" "chat-agent-system/internal/auth" "chat-agent-system/internal/authz" "chat-agent-system/internal/bizdev" @@ -45,6 +47,8 @@ type Server struct { instSvc *projectinstance.Service notifySvc *notification.Service resolver *workspace.Resolver + wsCap *workspace.Capability + cmds *taskrun.Commands httpServer *http.Server } @@ -93,6 +97,18 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { if pool != nil { taskMgr.SetStore(taskrun.NewPGStore(pool)) } + wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable()) + cmds := &taskrun.Commands{ + Ctx: svc.ContextManager(), + PW: pwSvc, + Env: cfg.Env, + StorageIdentity: "env:" + cfg.Env, + } + if pool != nil { + cmds.UoW = uow.NewPG(pool) + cmds.Fav = &artifactvault.Ingest{Env: cfg.Env} + } + taskMgr.SetCommands(cmds) s := &Server{ cfg: cfg, @@ -109,6 +125,8 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { instSvc: instSvc, notifySvc: notification.NewService(pool), resolver: workspace.NewResolver(cfg.DataDir, pool), + wsCap: wsCap, + cmds: cmds, } if instSvc != nil && instSvc.Available() { svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc)) diff --git a/internal/server/sse.go b/internal/server/sse.go index d19cbc8d..ffad0f19 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -334,7 +334,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { req.DeviceType = q.Get("device_type") req.IntendedUse = q.Get("intended_use") req.TargetMarket = q.Get("target_market") - req.Backend = q.Get("backend") + req.Backend = resolveTaskBackend(s, q.Get("backend")) req.WorkDir = q.Get("work_dir") req.AutoRepair, _ = strconv.ParseBool(q.Get("auto_repair")) if pid := q.Get("project_id"); pid != "" { diff --git a/internal/server/task_finalize.go b/internal/server/task_finalize.go index 1d933c43..1a84ed13 100644 --- a/internal/server/task_finalize.go +++ b/internal/server/task_finalize.go @@ -5,12 +5,11 @@ import ( "crypto/sha256" "encoding/hex" "fmt" - "os" - "path/filepath" "chat-agent-system/internal/bizdev" "chat-agent-system/internal/squad" "chat-agent-system/internal/taskrun" + "chat-agent-system/internal/workspace" ) // FinalizeTaskRunProjection 将任务执行结果投影到 workflow_runs 与聊天助手消息（供回填/管理命令使用）。 @@ -18,101 +17,69 @@ func (s *Server) FinalizeTaskRunProjection(ctx context.Context, record *taskrun. return s.finalizeTaskRun(ctx, record, result) } -// finalizeTaskRun 将任务执行结果投影到 workflow_runs 与聊天助手消息。 +// finalizeTaskRun 是薄适配：产物哈希在事务外经 workspace.Capability 读取，终态命令交给 taskrun.Commands。 func (s *Server) finalizeTaskRun(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) { if record == nil || result == nil { return "failed", fmt.Errorf("finalize: missing record or result") } + if result.DocRecord != nil && result.DocRecord.ContentSHA256 == "" { + if hexSum := s.hashDeliverable(ctx, record, result); hexSum != "" { + result.DocRecord.ContentSHA256 = hexSum + } + } + cmds := s.taskCommands() + if s != nil { + cmds.PW = s.pwSvc + if s.svc != nil { + cmds.Ctx = s.svc.ContextManager() + } + } + return cmds.Finalize(ctx, record, result) +} - workspaceID := result.WorkDir - if workspaceID == "" { - workspaceID = record.WorkspaceID +func (s *Server) taskCommands() *taskrun.Commands { + if s != nil && s.cmds != nil { + return s.cmds } + cmds := &taskrun.Commands{ + PW: s.pwSvc, + Env: "test", + } + if s != nil && s.cfg != nil { + cmds.Env = s.cfg.Env + cmds.StorageIdentity = "env:" + s.cfg.Env + } + if s != nil && s.svc != nil { + cmds.Ctx = s.svc.ContextManager() + } + return cmds +} - terminalStatus := result.TerminalStatus() - summaryContent := result.CompletionSummary() - - docID := "" - artifactTitle := result.ArtifactTitle - sha256Hex := "" - if result.DocRecord != nil { - docID = result.DocRecord.DocID - if result.DocRecord.ContentSHA256 != "" { - sha256Hex = result.DocRecord.ContentSHA256 - } +func (s *Server) hashDeliverable(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) string { + if s == nil || s.wsCap == nil || record == nil { + return "" } - if docID == "" && record.ProjectID > 0 && record.TaskKey != "" { - docID = fmt.Sprintf("doc_proj%d_%s", record.ProjectID, record.TaskKey) + id := workspace.OpaqueID(result.WorkDir, record.WorkspaceID) + if id == "" { + return "" } - if artifactTitle == "" && record.TaskKey != "" { - artifactTitle = squad.TaskArtifactTitle(record.TaskKey) + env := "" + if s.cfg != nil { + env = s.cfg.Env } - - if sha256Hex == "" && workspaceID != "" { - if computed, err := hashOutputFile(workspaceID); err == nil { - sha256Hex = computed - } + ref := workspace.WorkspaceRef{ + ID: id, + SchemaVersion: workspace.SchemaVersion, + ProjectScope: record.ProjectID, + RunID: record.RunID, + Environment: env, } - - meta := map[string]any{ - "run_id": record.RunID, - "workspace_id": workspaceID, - "task_status": terminalStatus, - "task_key": record.TaskKey, + body, err := s.wsCap.ReadArtifact(ctx, ref, workspace.DeliverableKey) + if err != nil || len(body) == 0 { + return "" } - if record.ProjectID > 0 { - meta["project_id"] = record.ProjectID - } - if art := result.ArtifactMetadata(); art != nil { - if workspaceID != "" { - art["workspace_id"] = workspaceID - } - if sha256Hex != "" { - art["sha256"] = sha256Hex - } - if artifactTitle != "" { - art["title"] = artifactTitle - } - if record.TaskKey != "" { - art["task_key"] = record.TaskKey - } - if docID != "" { - art["doc_id"] = docID - } - meta["artifact"] = art - } else if docID != "" { - artifact := map[string]any{ - "doc_id": docID, - "title": artifactTitle, - "task_key": record.TaskKey, - "workspace_id": workspaceID, - } - if sha256Hex != "" { - artifact["sha256"] = sha256Hex - } - 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 - } - } - meta["artifact"] = artifact - } - - if s.pwSvc != nil && record.ProjectID > 0 && record.RunID != "" { - if err := s.pwSvc.UpdateWorkflowRun(ctx, record.RunID, terminalStatus, docID, artifactTitle, ""); err != nil { - return "finalizing_failed", fmt.Errorf("update workflow run: %w", err) - } - } - - if s.svc != nil && s.svc.ContextManager() != nil && record.AssistantMsgID > 0 { - if err := s.svc.ContextManager().FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summaryContent, meta, terminalStatus); err != nil { - return "finalizing_failed", fmt.Errorf("finalize assistant message: %w", err) - } - } - - return terminalStatus, nil + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]) } func resolveTerminalStatus(result *bizdev.TaskRunResult) string { @@ -124,19 +91,6 @@ func buildTaskCompletionSummary(result *bizdev.TaskRunResult, terminalStatus str return result.CompletionSummary() } -func hashOutputFile(workDir string) (string, error) { - path := filepath.Join(workDir, "output.md") - data, err := os.ReadFile(path) - if err != nil { - return "", err - } - if len(data) == 0 { - return "", fmt.Errorf("output.md is empty") - } - sum := sha256.Sum256(data) - return hex.EncodeToString(sum[:]), nil -} - func squadResultToTaskRun(res *squad.SquadRunResult, taskKey string) *bizdev.TaskRunResult { if res == nil { return nil diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index 5a6e6b9e..a25babf6 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -645,7 +645,7 @@ func (s *Server) buildBizdevTaskRequest(dto TaskRunRequestDTO, sessionID, client TargetMarket: dto.TargetMarket, Fields: dto.Fields, Materials: dto.Materials, - Backend: dto.Backend, + Backend: resolveTaskBackend(s, dto.Backend), AutoRepair: true, // 产品契约：受控小队任务服务端启用修复，不依赖浏览器字段 SessionID: sessionID, UserID: dto.UserID, @@ -664,6 +664,19 @@ func (s *Server) buildBizdevTaskRequest(dto TaskRunRequestDTO, sessionID, client } } +func resolveTaskBackend(s *Server, requested string) string { + configured := "" + if s != nil && s.cfg != nil { + configured = s.cfg.AgentBackend + } + resolved, err := taskrun.ResolveBackend(requested, configured) + if err != nil { + fallback, _ := taskrun.ResolveBackend("", configured) + return fallback + } + return resolved +} + func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clientTurnID, runID string) bizdev.StageRunRequest { return bizdev.StageRunRequest{ Stage: bizdev.MaturityStage(dto.Stage), @@ -674,7 +687,7 @@ func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clien TargetMarket: dto.TargetMarket, Fields: dto.Fields, Materials: dto.Materials, - Backend: dto.Backend, + Backend: resolveTaskBackend(s, dto.Backend), AutoRepair: true, // 产品契约：阶段任务同样启用修复预算 SessionID: sessionID, UserID: dto.UserID, diff --git a/internal/workspace/cleanup.go b/internal/workspace/cleanup.go new file mode 100644 index 00000000..57a49c7f --- /dev/null +++ b/internal/workspace/cleanup.go @@ -0,0 +1,111 @@ +package workspace + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" +) + +// OpaqueID turns a stored work dir or already-opaque id into a WorkspaceRef.ID. +// It never returns a physical path. +func OpaqueID(parts ...string) string { + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + continue + } + slash := filepath.ToSlash(p) + if i := strings.LastIndex(slash, "/workspaces/"); i >= 0 { + rest := slash[i+len("/workspaces/"):] + if rest == "" { + continue + } + return strings.Split(rest, "/")[0] + } + if !strings.Contains(slash, "/") { + return p + } + return filepath.Base(filepath.Clean(p)) + } + return "" +} + +type RetentionReceipt struct { + LogicalID string + Digest string + StorageIdentity string + Verified bool +} + +type FrozenManifest struct { + SHA256 string + StorageIdentity string + WorkspaceID string + Members []FrozenMember +} + +type FrozenMember struct { + LogicalID string + Digest string + Kind string +} + +func memberNeedsRetention(kind string) bool { + switch kind { + case "document", "assistant_turn", "run_manifest", "attachment": + return true + default: + return false + } +} + +var ErrCleanupBlocked = fmt.Errorf("workspace: cleanup blocked until verified receipts match the frozen manifest") + +func AssertCleanupAllowed(frozen FrozenManifest, receipts []RetentionReceipt) error { + if frozen.SHA256 == "" || frozen.StorageIdentity == "" { + return ErrCleanupBlocked + } + byLogical := map[string]RetentionReceipt{} + for _, r := range receipts { + byLogical[r.LogicalID] = r + } + for _, mem := range frozen.Members { + if !memberNeedsRetention(mem.Kind) { + continue + } + r, ok := byLogical[mem.LogicalID] + if !ok || !r.Verified { + return ErrCleanupBlocked + } + if r.Digest != mem.Digest || r.StorageIdentity != frozen.StorageIdentity { + return ErrCleanupBlocked + } + } + return nil +} + +// Cleanup removes the workspace directory only after verified receipts match the frozen manifest. +func (c *Capability) Cleanup(ctx context.Context, ref WorkspaceRef, frozen FrozenManifest, receipts []RetentionReceipt) error { + if err := AssertCleanupAllowed(frozen, receipts); err != nil { + return err + } + if err := c.guard(ctx, ref, false); err != nil { + return err + } + root := filepath.Join(c.dataRoot, "workspaces", ref.ID) + abs, err := filepath.Abs(root) + if err != nil { + return err + } + dataRoot, err := filepath.Abs(c.dataRoot) + if err != nil { + return err + } + rel, err := filepath.Rel(dataRoot, abs) + if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) { + return ErrPathEscape + } + return os.RemoveAll(abs) +} diff --git a/internal/workspace/cleanup_test.go b/internal/workspace/cleanup_test.go new file mode 100644 index 00000000..381fa673 --- /dev/null +++ b/internal/workspace/cleanup_test.go @@ -0,0 +1,63 @@ +package workspace + +import ( + "context" + "os" + "path/filepath" + "testing" +) + +func TestOpaqueID_StripsPhysicalPath(t *testing.T) { + got := OpaqueID("/data/dev/workspaces/run_abc/output.md", "") + if got != "run_abc" { + t.Fatalf("got %q", got) + } + if OpaqueID("already-opaque") != "already-opaque" { + t.Fatal("opaque id changed") + } +} + +func TestCleanup_RequiresVerifiedReceipts(t *testing.T) { + root := t.TempDir() + cap := NewCapability(root, nil) + ctx := context.Background() + ref, err := cap.Issue(WorkspaceRef{ID: "ws_clean", ProjectScope: 1, RunID: "run_clean"}) + if err != nil { + t.Fatal(err) + } + if err := cap.WriteArtifact(ctx, ref, DeliverableKey, []byte("# body\n")); err != nil { + t.Fatal(err) + } + frozen := FrozenManifest{ + SHA256: "deadbeef", StorageIdentity: "env:dev", WorkspaceID: ref.ID, + Members: []FrozenMember{{LogicalID: "doc_a", Digest: "abc", Kind: "document"}}, + } + if err := cap.Cleanup(ctx, ref, frozen, nil); err == nil { + t.Fatal("cleanup without receipts must fail") + } + if _, err := os.Stat(filepath.Join(root, "workspaces", ref.ID, DeliverableKey)); err != nil { + t.Fatalf("workspace removed without receipts: %v", err) + } + receipts := []RetentionReceipt{{ + LogicalID: "doc_a", Digest: "abc", StorageIdentity: "env:dev", Verified: true, + }} + if err := cap.Cleanup(ctx, ref, frozen, receipts); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(root, "workspaces", ref.ID)); !os.IsNotExist(err) { + t.Fatalf("workspace still present: %v", err) + } +} + +func TestCleanup_MismatchDigestRejected(t *testing.T) { + frozen := FrozenManifest{ + SHA256: "x", StorageIdentity: "env:dev", + Members: []FrozenMember{{LogicalID: "doc_a", Digest: "abc", Kind: "document"}}, + } + err := AssertCleanupAllowed(frozen, []RetentionReceipt{{ + LogicalID: "doc_a", Digest: "other", StorageIdentity: "env:dev", Verified: true, + }}) + if err == nil { + t.Fatal("mismatch must fail") + } +} diff --git a/internal/workspace/keys.go b/internal/workspace/keys.go new file mode 100644 index 00000000..abc4e8e2 --- /dev/null +++ b/internal/workspace/keys.go @@ -0,0 +1,4 @@ +package workspace + +// DeliverableKey is the only artifact key for the task run markdown deliverable. +const DeliverableKey = "output.md" From 3997a04bdb3896e079216515c450a77f2e29443a Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 02:06:21 +0000 Subject: [PATCH 5/7] test(taskrun): cover duplicate finalize, cancel race, and SSE Use an isolated dual-pool database so idempotent finalize, cancel CAS, commit-fail SSE, and receipt-gated cleanup are proven without a process mutex. Co-authored-by: Cursor --- internal/server/task_finalize_test.go | 52 ++++ internal/taskrun/finalize_pg_test.go | 363 ++++++++++++++++++++++++++ internal/taskrun/finalize_test.go | 131 ++++++++++ internal/taskrun/manifest_test.go | 77 ++++++ 4 files changed, 623 insertions(+) create mode 100644 internal/taskrun/finalize_pg_test.go create mode 100644 internal/taskrun/finalize_test.go create mode 100644 internal/taskrun/manifest_test.go diff --git a/internal/server/task_finalize_test.go b/internal/server/task_finalize_test.go index 37c65bc4..12d21518 100644 --- a/internal/server/task_finalize_test.go +++ b/internal/server/task_finalize_test.go @@ -2,6 +2,8 @@ package server import ( "context" + "crypto/sha256" + "encoding/hex" "os" "path/filepath" "strings" @@ -15,6 +17,7 @@ import ( "chat-agent-system/internal/squad" "chat-agent-system/internal/stream" "chat-agent-system/internal/taskrun" + "chat-agent-system/internal/workspace" ) func TestFinalizeTaskRun_ReplacesPlaceholderContent(t *testing.T) { @@ -298,3 +301,52 @@ func TestAsyncTaskRun_PendingConfirmation_FinalizesAsPending(t *testing.T) { } t.Fatal("worker did not finish") } + +func TestFinalizeTaskRun_HashesViaWorkspaceCapability(t *testing.T) { + cfg := &config.Config{Env: "test", DataDir: t.TempDir(), AgentBackend: "stub"} + svc := bizdev.MustNewService(t, cfg) + srv := NewServer(cfg, svc) + ctx := context.Background() + ref, err := srv.wsCap.Issue(workspace.WorkspaceRef{ID: "ws_hash", ProjectScope: 1, RunID: "run_hash"}) + if err != nil { + t.Fatal(err) + } + body := []byte("# hashed via capability\n") + if err := srv.wsCap.WriteArtifact(ctx, ref, workspace.DeliverableKey, body); err != nil { + t.Fatal(err) + } + ctxMgr := svc.ContextManager() + asstID, err := ctxMgr.RecordUserMessageWithID(ctx, "sess-hash", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil) + if err != nil { + t.Fatal(err) + } + pwSvc := projectworkflow.NewService(projectworkflow.NewStore(nil), nil) + _, _ = pwSvc.CreateWorkflowRun(ctx, &projectworkflow.WorkflowRun{ + RunID: "run_hash", ProjectID: 1, TaskKey: "intake", Status: "running", + }) + srv.pwSvc = pwSvc + record := &taskrun.RunRecord{RunID: "run_hash", SessionID: "sess-hash", ProjectID: 1, TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: ref.ID} + status, err := srv.finalizeTaskRun(ctx, record, &bizdev.TaskRunResult{ + Success: true, TaskKey: "intake", WorkDir: ref.ID, + DocRecord: &docgov.DocumentRecord{DocID: "doc_hash"}, + }) + if err != nil { + t.Fatal(err) + } + if status != "completed" { + t.Fatalf("status=%s", status) + } + msg, err := ctxMgr.GetMessageByID(ctx, asstID) + if err != nil { + t.Fatal(err) + } + art, _ := msg.Metadata["artifact"].(map[string]any) + if art == nil { + t.Fatalf("artifact missing: %#v", msg.Metadata) + } + sum := sha256.Sum256(body) + want := hex.EncodeToString(sum[:]) + if art["sha256"] != want { + t.Fatalf("sha256=%v want %s", art["sha256"], want) + } +} diff --git a/internal/taskrun/finalize_pg_test.go b/internal/taskrun/finalize_pg_test.go new file mode 100644 index 00000000..de352d73 --- /dev/null +++ b/internal/taskrun/finalize_pg_test.go @@ -0,0 +1,363 @@ +package taskrun_test + +import ( + "context" + "encoding/json" + "errors" + "net/url" + "os" + "path/filepath" + "sync" + "testing" + "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/db" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/stream" + "chat-agent-system/internal/taskrun" + "chat-agent-system/internal/workspace" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func init() { + if _, err := os.Stat("deploy/postgres/goose"); err == nil { + return + } + for _, cand := range []string{"../..", "../../.."} { + if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() { + _ = os.Chdir(cand) + return + } + } +} + +var ( + isoCleanup func() +) + +func TestMain(m *testing.M) { + code := m.Run() + if isoCleanup != nil { + isoCleanup() + } + os.Exit(code) +} + +func rewriteDatabaseName(dsn, name string) string { + if u, err := url.Parse(dsn); err == nil && u.Scheme != "" { + u.Path = "/" + name + return u.String() + } + return dsn +} + +var isoOnce sync.Once +var isoPoolA, isoPoolB *pgxpool.Pool +var isoErr error + +func openIsolatedM3(t *testing.T) (*pgxpool.Pool, *pgxpool.Pool) { + t.Helper() + if testing.Short() { + t.Skip("U8 PG races are not part of -short") + } + dsn := db.SkipIfNoTestDatabaseURL(t) + db.RefuseSharedEnvDSN(t, dsn) + + isoOnce.Do(func() { + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + adminDSN := rewriteDatabaseName(dsn, "postgres") + admin, err := pgxpool.New(ctx, adminDSN) + if err != nil { + isoErr = err + return + } + const isolated = "agent_rel_m3_u8" + _, _ = admin.Exec(ctx, `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=$1 AND pid <> pg_backend_pid()`, isolated) + _, _ = admin.Exec(ctx, `DROP DATABASE IF EXISTS `+isolated) + if _, err := admin.Exec(ctx, `CREATE DATABASE `+isolated); err != nil { + admin.Close() + isoErr = err + return + } + isoDSN := rewriteDatabaseName(dsn, isolated) + poolA, err := pgxpool.New(ctx, isoDSN) + if err != nil { + admin.Close() + isoErr = err + return + } + if err := db.RunMigrations(ctx, poolA, db.ResolveMigrationsDir()); err != nil { + poolA.Close() + admin.Close() + isoErr = err + return + } + poolB, err := pgxpool.New(ctx, isoDSN) + if err != nil { + poolA.Close() + admin.Close() + isoErr = err + return + } + isoPoolA, isoPoolB = poolA, poolB + isoCleanup = func() { + poolA.Close() + poolB.Close() + _, _ = admin.Exec(context.Background(), `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=$1 AND pid <> pg_backend_pid()`, isolated) + _, _ = admin.Exec(context.Background(), `DROP DATABASE IF EXISTS `+isolated) + admin.Close() + } + }) + if isoErr != nil { + t.Fatalf("isolated db: %v", isoErr) + } + return isoPoolA, isoPoolB +} + +func seedUserAndProject(t *testing.T, pool *pgxpool.Pool) (userID, projectID int64) { + t.Helper() + ctx := context.Background() + err := pool.QueryRow(ctx, `SELECT id FROM users ORDER BY id LIMIT 1`).Scan(&userID) + if err != nil { + t.Fatalf("seed user: %v", err) + } + err = pool.QueryRow(ctx, `SELECT id FROM projects ORDER BY id LIMIT 1`).Scan(&projectID) + if err != nil { + t.Fatalf("seed project: %v", err) + } + return userID, projectID +} + +func insertRunning(t *testing.T, pool *pgxpool.Pool, rec *taskrun.RunRecord) { + t.Helper() + _, err := pool.Exec(context.Background(), ` + INSERT INTO task_runs ( + run_id, session_id, project_id, user_id, client_turn_id, task_key, status, + workspace_id, assistant_message_id, created_at, updated_at + ) VALUES ($1,$2,$3,$4,$5,$6,'running',$7,$8,now(),now()) + `, rec.RunID, rec.SessionID, rec.ProjectID, rec.UserID, rec.ClientTurnID, rec.TaskKey, rec.WorkspaceID, rec.AssistantMsgID) + if err != nil { + t.Fatalf("insert run: %v", err) + } +} + +func TestPG_DuplicateFinalizeIdempotent(t *testing.T) { + poolA, poolB := openIsolatedM3(t) + userID, projectID := seedUserAndProject(t, poolA) + ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{}) + sessionID := "sess_u8_dup" + asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID}) + if err != nil { + t.Fatal(err) + } + rec := &taskrun.RunRecord{ + RunID: "run_u8_dup", SessionID: sessionID, ProjectID: projectID, UserID: userID, + TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: "ws_u8_dup", + } + insertRunning(t, poolA, rec) + body := []byte("# committed intake body\n") + res := &bizdev.TaskRunResult{ + Success: true, TaskKey: "intake", ArtifactTitle: "用户初始需求收集表", MarkdownBody: string(body), + DocRecord: &docgov.DocumentRecord{DocID: "doc_u8_dup", ContentSHA256: artifactvault.SHA256Hex(body), Version: 1, ProjectID: projectID}, + } + cmdsA := &taskrun.Commands{UoW: uow.NewPG(poolA), Ctx: ctxMgr, Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"} + cmdsB := &taskrun.Commands{UoW: uow.NewPG(poolB), Ctx: ctxmgr.NewContextManager(poolB, nil, ctxmgr.TimeDecayConfig{}), Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"} + st1, err := cmdsA.Finalize(context.Background(), rec, res) + if err != nil { + t.Fatal(err) + } + st2, err := cmdsB.Finalize(context.Background(), rec, res) + if err != nil { + t.Fatal(err) + } + if st1 != "completed" || st2 != "completed" { + t.Fatalf("status %s %s", st1, st2) + } + var msgs, mans, envs, stgs int + _ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM chat_messages WHERE session_id=$1 AND role='assistant'`, sessionID).Scan(&msgs) + _ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM task_run_manifests WHERE run_id=$1`, rec.RunID).Scan(&mans) + _ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM event_envelopes WHERE aggregate_id=$1`, rec.RunID).Scan(&envs) + _ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM artifact_payload_staging WHERE source_event_id=$1`, "taskrun:"+rec.RunID+":completed").Scan(&stgs) + if msgs != 1 || mans != 1 || envs != 1 || stgs != 1 { + t.Fatalf("counts messages=%d manifests=%d envelopes=%d staging=%d", msgs, mans, envs, stgs) + } + var sha1, sha2 string + _ = poolA.QueryRow(context.Background(), `SELECT sha256 FROM task_run_manifests WHERE run_id=$1`, rec.RunID).Scan(&sha1) + _ = poolB.QueryRow(context.Background(), `SELECT sha256 FROM task_run_manifests WHERE run_id=$1`, rec.RunID).Scan(&sha2) + if sha1 == "" || sha1 != sha2 { + t.Fatalf("manifest sha %q %q", sha1, sha2) + } +} + +func TestPG_FinalizeCancelRaceOneTerminal(t *testing.T) { + poolA, poolB := openIsolatedM3(t) + userID, projectID := seedUserAndProject(t, poolA) + ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{}) + sessionID := "sess_u8_race" + asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID}) + if err != nil { + t.Fatal(err) + } + rec := &taskrun.RunRecord{ + RunID: "run_u8_race", SessionID: sessionID, ProjectID: projectID, UserID: userID, + TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: "ws_u8_race", + } + insertRunning(t, poolA, rec) + body := []byte("# race body\n") + res := &bizdev.TaskRunResult{ + Success: true, TaskKey: "intake", MarkdownBody: string(body), + DocRecord: &docgov.DocumentRecord{DocID: "doc_u8_race", ContentSHA256: artifactvault.SHA256Hex(body), Version: 1}, + } + cmdsA := &taskrun.Commands{UoW: uow.NewPG(poolA), Ctx: ctxMgr, Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"} + cmdsB := &taskrun.Commands{UoW: uow.NewPG(poolB), Ctx: ctxmgr.NewContextManager(poolB, nil, ctxmgr.TimeDecayConfig{}), Env: "dev", StorageIdentity: "env:dev"} + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _, _ = cmdsA.Finalize(context.Background(), rec, res) + }() + go func() { + defer wg.Done() + _, _ = cmdsB.Cancel(context.Background(), rec) + }() + wg.Wait() + var status string + if err := poolA.QueryRow(context.Background(), `SELECT status FROM task_runs WHERE run_id=$1`, rec.RunID).Scan(&status); err != nil { + t.Fatal(err) + } + if status != "completed" && status != "cancelled" { + t.Fatalf("illegal terminal %q", status) + } + var n int + _ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM chat_messages WHERE session_id=$1 AND role='assistant'`, sessionID).Scan(&n) + if n != 1 { + t.Fatalf("assistant turns=%d", n) + } +} + +func TestPG_SSEDoneNotEmittedWhenCommitFails(t *testing.T) { + poolA, _ := openIsolatedM3(t) + userID, projectID := seedUserAndProject(t, poolA) + ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{}) + sessionID := "sess_u8_sse" + asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID}) + if err != nil { + t.Fatal(err) + } + rec := &taskrun.RunRecord{ + RunID: "run_u8_sse", SessionID: sessionID, ProjectID: projectID, UserID: userID, + TaskKey: "intake", AssistantMsgID: asstID, Status: "pending", + } + unit := uow.NewPG(poolA) + unit.BeforeCommit = func(uow.TxScope) error { return errors.New("inject commit fail") } + cmds := &taskrun.Commands{UoW: unit, Ctx: ctxMgr, Env: "dev", StorageIdentity: "env:dev"} + mgr := taskrun.NewManager(nil, ctxMgr) + mgr.SetStore(taskrun.NewPGStore(poolA)) + mgr.SetCommands(cmds) + mgr.SetExecutor(func(context.Context, string, bizdev.TaskRunRequest, *stream.Streamer) (*bizdev.TaskRunResult, error) { + return &bizdev.TaskRunResult{Success: true, MarkdownBody: "# sse\n"}, nil + }, nil, nil) + if _, err := mgr.CreateAndStart(context.Background(), rec.RunID, rec, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, ""); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(8 * time.Second) + for time.Now().Before(deadline) { + got, err := mgr.GetRun(rec.RunID) + if err == nil && got.Status == "finalizing_failed" { + evs, listErr := taskrun.NewPGStore(poolA).ListEvents(context.Background(), rec.RunID, 0) + if listErr != nil { + t.Fatal(listErr) + } + for _, ev := range evs { + if ev.Event.Type == stream.EventDone { + t.Fatal("done emitted on failed commit") + } + } + msg, _ := ctxMgr.GetMessageByID(context.Background(), asstID) + if msg != nil && msg.Content != "任务执行中…" { + t.Fatalf("placeholder mutated after rollback: %q", msg.Content) + } + return + } + time.Sleep(30 * time.Millisecond) + } + t.Fatal("expected finalizing_failed") +} + +func TestPG_CleanupWithoutReceiptFailsAndStagingSurvives(t *testing.T) { + poolA, _ := openIsolatedM3(t) + userID, projectID := seedUserAndProject(t, poolA) + ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{}) + sessionID := "sess_u8_clean" + asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID}) + if err != nil { + t.Fatal(err) + } + dataRoot := t.TempDir() + cap := workspace.NewCapability(dataRoot, nil) + ref, err := cap.Issue(workspace.WorkspaceRef{ID: "ws_u8_clean", ProjectScope: projectID, RunID: "run_u8_clean"}) + if err != nil { + t.Fatal(err) + } + body := []byte("# vaulted body for cleanup\n") + if err := cap.WriteArtifact(context.Background(), ref, workspace.DeliverableKey, body); err != nil { + t.Fatal(err) + } + rec := &taskrun.RunRecord{ + RunID: "run_u8_clean", SessionID: sessionID, ProjectID: projectID, UserID: userID, + TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: ref.ID, + } + insertRunning(t, poolA, rec) + res := &bizdev.TaskRunResult{ + Success: true, TaskKey: "intake", MarkdownBody: string(body), WorkDir: ref.ID, + DocRecord: &docgov.DocumentRecord{DocID: "doc_u8_clean", ContentSHA256: artifactvault.SHA256Hex(body), Version: 2}, + } + cmds := &taskrun.Commands{UoW: uow.NewPG(poolA), Ctx: ctxMgr, Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"} + if _, err := cmds.Finalize(context.Background(), rec, res); err != nil { + t.Fatal(err) + } + var manSHA, storageID string + var members []byte + if err := poolA.QueryRow(context.Background(), `SELECT sha256, storage_identity, members FROM task_run_manifests WHERE run_id=$1`, rec.RunID). + Scan(&manSHA, &storageID, &members); err != nil { + t.Fatal(err) + } + var parsed []taskrun.ManifestMember + if err := json.Unmarshal(members, &parsed); err != nil { + t.Fatal(err) + } + frozen := workspace.FrozenManifest{SHA256: manSHA, StorageIdentity: storageID, WorkspaceID: ref.ID} + for _, m := range parsed { + frozen.Members = append(frozen.Members, workspace.FrozenMember{LogicalID: m.LogicalID, Digest: m.Digest, Kind: m.Kind}) + } + if err := cap.Cleanup(context.Background(), ref, frozen, nil); err == nil { + t.Fatal("cleanup without receipts must fail") + } + if _, err := os.Stat(filepath.Join(dataRoot, "workspaces", ref.ID, workspace.DeliverableKey)); err != nil { + t.Fatalf("workspace removed without receipts: %v", err) + } + var receipts []workspace.RetentionReceipt + for _, m := range parsed { + receipts = append(receipts, workspace.RetentionReceipt{ + LogicalID: m.LogicalID, Digest: m.Digest, StorageIdentity: storageID, Verified: true, + }) + } + if err := cap.Cleanup(context.Background(), ref, frozen, receipts); err != nil { + t.Fatal(err) + } + eventID := "taskrun:" + rec.RunID + ":completed" + reg := artifactvault.NewPGRegistry(poolA) + st, err := reg.LoadStaging(context.Background(), eventID) + if err != nil { + t.Fatal(err) + } + if artifactvault.SHA256Hex(st.Payload) != artifactvault.SHA256Hex(body) { + t.Fatalf("staging sha drifted") + } +} diff --git a/internal/taskrun/finalize_test.go b/internal/taskrun/finalize_test.go new file mode 100644 index 00000000..210495d6 --- /dev/null +++ b/internal/taskrun/finalize_test.go @@ -0,0 +1,131 @@ +package taskrun_test + +import ( + "context" + "strings" + "sync" + "testing" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/taskrun" +) + +func TestCommands_DuplicateFinalizeMemory(t *testing.T) { + ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{}) + asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-dup", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"task_status": "running"}) + if err != nil { + t.Fatal(err) + } + cmds := &taskrun.Commands{Ctx: ctxMgr, Env: "test", StorageIdentity: "env:test"} + rec := &taskrun.RunRecord{RunID: "run_dup_mem", SessionID: "sess-dup", TaskKey: "intake", AssistantMsgID: asstID} + res := &bizdev.TaskRunResult{ + Success: true, TaskKey: "intake", ArtifactTitle: "用户初始需求收集表", MarkdownBody: "# body\n", + DocRecord: &docgov.DocumentRecord{DocID: "doc_dup", ContentSHA256: strings.Repeat("a", 64), Version: 1}, + } + st1, err := cmds.Finalize(context.Background(), rec, res) + if err != nil { + t.Fatal(err) + } + msg, err := ctxMgr.GetMessageByID(context.Background(), asstID) + if err != nil { + t.Fatal(err) + } + if msg.Content == "任务执行中…" { + t.Fatal("placeholder not replaced") + } + st2, err := cmds.Finalize(context.Background(), rec, res) + if err != nil { + t.Fatal(err) + } + if st1 != "completed" || st2 != "completed" { + t.Fatalf("status %s %s", st1, st2) + } + msg2, _ := ctxMgr.GetMessageByID(context.Background(), asstID) + if msg2.ID != msg.ID || msg2.Content != msg.Content { + t.Fatal("duplicate finalize mutated the assistant turn") + } +} + +func TestCommands_CancelWinsOverFinalizeMemory(t *testing.T) { + ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{}) + asstID, _ := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-race", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil) + cmds := &taskrun.Commands{Ctx: ctxMgr, Env: "test", StorageIdentity: "env:test"} + rec := &taskrun.RunRecord{RunID: "run_race_mem", SessionID: "sess-race", AssistantMsgID: asstID} + st, err := cmds.Cancel(context.Background(), rec) + if err != nil { + t.Fatal(err) + } + if st != taskrun.StatusCancelled { + t.Fatalf("status=%s", st) + } + res := &bizdev.TaskRunResult{Success: true, MarkdownBody: "# later\n"} + st2, err := cmds.Finalize(context.Background(), rec, res) + if err != nil { + t.Fatal(err) + } + if st2 != taskrun.StatusCancelled { + t.Fatalf("finalize must adopt cancelled, got %s", st2) + } +} + +func TestCommands_ConcurrentMemorySingleTerminal(t *testing.T) { + ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{}) + asstID, _ := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-conc", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil) + cmds := &taskrun.Commands{Ctx: ctxMgr, Env: "test", StorageIdentity: "env:test"} + rec := &taskrun.RunRecord{RunID: "run_conc_mem", SessionID: "sess-conc", AssistantMsgID: asstID} + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _, _ = cmds.Finalize(context.Background(), rec, &bizdev.TaskRunResult{Success: true, MarkdownBody: "# ok\n"}) + }() + go func() { + defer wg.Done() + _, _ = cmds.Cancel(context.Background(), rec) + }() + wg.Wait() + msg, err := ctxMgr.GetMessageByID(context.Background(), asstID) + if err != nil { + t.Fatal(err) + } + if msg.ID != asstID { + t.Fatalf("assistant id changed %d", msg.ID) + } +} + +func TestRunManifest_CleanupGate(t *testing.T) { + man := &taskrun.RunManifest{ + RunID: "run_clean", WorkspaceID: "ws", StorageIdentity: "env:dev", + Members: []taskrun.ManifestMember{{LogicalID: "doc_a", Digest: "abc", Kind: taskrun.MemberKindDocument}}, + } + if err := man.Freeze(); err != nil { + t.Fatal(err) + } + if err := man.AssertCleanupAllowed(nil); err == nil { + t.Fatal("cleanup without receipts must fail") + } + receipts := []taskrun.VerifiedReceipt{{ + LogicalID: "doc_a", Digest: "abc", StorageIdentity: "env:dev", Verified: true, + }} + if err := man.AssertCleanupAllowed(receipts); err != nil { + t.Fatal(err) + } + bad := []taskrun.VerifiedReceipt{{ + LogicalID: "doc_a", Digest: "zzz", StorageIdentity: "env:dev", Verified: true, + }} + if err := man.AssertCleanupAllowed(bad); err == nil { + t.Fatal("digest mismatch must fail") + } +} + +func TestCommands_RejectsSensitiveMarkdown(t *testing.T) { + cmds := &taskrun.Commands{Env: "test", StorageIdentity: "env:test"} + _, err := cmds.Finalize(context.Background(), &taskrun.RunRecord{RunID: "run_secret"}, &bizdev.TaskRunResult{ + Success: true, MarkdownBody: "api_key=super-secret", + }) + if err == nil { + t.Fatal("expected sensitive reject") + } +} diff --git a/internal/taskrun/manifest_test.go b/internal/taskrun/manifest_test.go new file mode 100644 index 00000000..a3657f22 --- /dev/null +++ b/internal/taskrun/manifest_test.go @@ -0,0 +1,77 @@ +package taskrun + +import ( + "encoding/json" + "testing" +) + +func TestRunManifest_FreezeStableSHA(t *testing.T) { + a := &RunManifest{ + RunID: "run_a", + WorkspaceID: "ws1", + StorageIdentity: "env:dev", + Members: []ManifestMember{ + {LogicalID: "doc_b", Digest: "bbb", Kind: MemberKindDocument}, + {LogicalID: "doc_a", Digest: "aaa", Kind: MemberKindDocument}, + }, + Extensions: map[string]json.RawMessage{ + "fav/policy": json.RawMessage(`{"v":1}`), + "ext/other": json.RawMessage(`{"k":"v"}`), + }, + } + b := &RunManifest{ + RunID: "run_a", + WorkspaceID: "ws1", + StorageIdentity: "env:dev", + Members: []ManifestMember{ + {LogicalID: "doc_a", Digest: "AAA", Kind: MemberKindDocument}, + {LogicalID: "doc_b", Digest: "bbb", Kind: MemberKindDocument}, + }, + Extensions: map[string]json.RawMessage{ + "ext/other": json.RawMessage(`{"k":"v"}`), + "fav/policy": json.RawMessage(`{"v":1}`), + }, + } + if err := a.Freeze(); err != nil { + t.Fatal(err) + } + if err := b.Freeze(); err != nil { + t.Fatal(err) + } + if a.SHA256 != b.SHA256 { + t.Fatalf("canonical sha diverged: %s vs %s\n%s\n%s", a.SHA256, b.SHA256, a.CanonicalBytes, b.CanonicalBytes) + } + if len(a.CanonicalBytes) == 0 || a.Members[0].LogicalID != "doc_a" { + t.Fatalf("members not sorted: %+v", a.Members) + } +} + +func TestRunManifest_RejectsIncompleteMember(t *testing.T) { + m := &RunManifest{RunID: "run_x", Members: []ManifestMember{{LogicalID: "x", Kind: MemberKindDocument}}} + if err := m.Freeze(); err == nil { + t.Fatal("expected incomplete member error") + } +} + +func TestResolveBackend_StripsUnauthorized(t *testing.T) { + if _, err := ResolveBackend("evil-shell", "stub"); err == nil { + t.Fatal("expected unauthorized") + } + got, err := ResolveBackend("", "pi-agent") + if err != nil || got != "pi-agent" { + t.Fatalf("got %q %v", got, err) + } +} + +func TestStripSecretsAndSensitive(t *testing.T) { + meta := StripSecrets(map[string]any{"api_key": "secret", "run_id": "r1", "authorization": "Bearer x"}) + if _, ok := meta["api_key"]; ok { + t.Fatal("api_key leaked") + } + if meta["run_id"] != "r1" { + t.Fatalf("run_id stripped: %#v", meta) + } + if err := rejectSensitiveText("hello api_key=abcd"); err == nil { + t.Fatal("expected sensitive reject") + } +} From ac302fb78f3c4685b8323f8512f34361008dc539 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 02:07:42 +0000 Subject: [PATCH 6/7] docs(ops): occupy the integration slot for M3 U8 merge Serialize merge and container rebuild so other agents wait while RunManifest lands on dev. Co-authored-by: Cursor --- .doc/ops/CURRENT.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index cc24229d..331672da 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -61,7 +61,7 @@ | holder | phase | since | scope | queue | |--------|-------|-------|-------|-------| -| - | idle | - | - | - | +| Cursor | merging | 2026-09-15 | M3 U8 feature/taskrun-runmanifest | - | **phase**: `idle` | `merging` | `smoke` | `e2e` From f4dfada9aea47d1c51cfedb1aa678282b896f81f Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 02:17:37 +0000 Subject: [PATCH 7/7] =?UTF-8?q?docs(test):=20M3=20U8=20TaskRun=20RunManife?= =?UTF-8?q?st=20=E9=AA=8C=E6=94=B6=E6=8A=A5=E5=91=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 隔离库双 pool、目标包 race、arch-check、web 与 rebuild 证据齐全；真实 PI 未测。最后一句：继续 U9。 Co-authored-by: Cursor --- .../ops}/2026-09-15-taskrun-runmanifest-u8.md | 22 ++-- .doc/ops/CURRENT.md | 11 +- ...4-module-decoupling-refactor-acceptance.md | 12 +- .doc/remediation/gap-matrix.md | 2 +- .doc/status/FEATURE-MATRIX.md | 4 +- ...9-15-taskrun-runmanifest-u8-test-report.md | 124 ++++++++++++++++++ .../arch-check-warn.txt | 4 + .../go-test-short.txt | 47 +++++++ .../go-test-u8-packages-race.txt | 4 + .../health.json | 1 + .../ready.json | 1 + .../rebuild-health-ready.txt | 4 + 12 files changed, 211 insertions(+), 25 deletions(-) rename .doc/{ops/short-term => archive/ops}/2026-09-15-taskrun-runmanifest-u8.md (73%) create mode 100644 .doc/test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md create mode 100644 .doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/arch-check-warn.txt create mode 100644 .doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-short.txt create mode 100644 .doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-u8-packages-race.txt create mode 100644 .doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/health.json create mode 100644 .doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/ready.json create mode 100644 .doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/rebuild-health-ready.txt diff --git a/.doc/ops/short-term/2026-09-15-taskrun-runmanifest-u8.md b/.doc/archive/ops/2026-09-15-taskrun-runmanifest-u8.md similarity index 73% rename from .doc/ops/short-term/2026-09-15-taskrun-runmanifest-u8.md rename to .doc/archive/ops/2026-09-15-taskrun-runmanifest-u8.md index a623aaae..0a36da1d 100644 --- a/.doc/ops/short-term/2026-09-15-taskrun-runmanifest-u8.md +++ b/.doc/archive/ops/2026-09-15-taskrun-runmanifest-u8.md @@ -1,6 +1,6 @@ --- title: "M3 U8 TaskRun 终态与 RunManifest" -status: in_progress +status: done created: 2026-09-15 updated: 2026-09-15 owner: Cursor @@ -35,25 +35,25 @@ TaskRun 终态在同一 UoW 内 CAS 写入唯一合法状态，并持久化类 ## 步骤 1. [x] CURRENT 认领 `mutex=taskrun`；从本地 `dev` 开 `feature/taskrun-runmanifest` -2. [ ] Goose 00037 Up+Down；`module-boundaries.yaml` 表属主 -3. [ ] Finalize 同 UoW：CAS + 助手 turn + manifest + envelope/staging -4. [ ] 清理门禁与 server 去路径、BackendRef -5. [ ] 隔离库双 pool / race；占槽 rebuild；报告最后一句只能「继续 U9」或「停止扩面」 +2. [x] Goose 00037 Up+Down；`module-boundaries.yaml` 表属主 +3. [x] Finalize 同 UoW：CAS + 助手 turn + manifest + envelope/staging +4. [x] 清理门禁与 server 去路径、BackendRef +5. [x] 隔离库双 pool / race；占槽 rebuild；报告最后一句只能「继续 U9」或「停止扩面」 ## 验收标准 -- [ ] 重复 finalize：同一 status、同一 manifest SHA、助手消息 1 条、envelope/staging 不加倍 -- [ ] finalize ∥ cancel：恰好一个合法终态 -- [ ] SSE `done` 不出现在 TX 失败路径；commit 前断线重试不第二 turn -- [ ] 无 verified receipt 则 cleanup FAIL；有 receipt 则可清 workspace 且 staging 仍能取出原 SHA -- [ ] 报告最后一句为「继续 U9」或「停止扩面」；U9 清单项保持未勾 +- [x] 重复 finalize：同一 status、同一 manifest SHA、助手消息 1 条、envelope/staging 不加倍 +- [x] finalize ∥ cancel：恰好一个合法终态 +- [x] SSE `done` 不出现在 TX 失败路径；commit 前断线重试不第二 turn +- [x] 无 verified receipt 则 cleanup FAIL；有 receipt 则可清 workspace 且 staging 仍能取出原 SHA +- [x] 报告最后一句为「继续 U9」或「停止扩面」；U9 清单项保持未勾 ## 相关链接 - Checklist: [2026-09-14-module-decoupling-refactor-acceptance.md](../checklists/2026-09-14-module-decoupling-refactor-acceptance.md) - Remediation: [2026-09-14-module-decoupling-refactor-plan.md](../../remediation/2026-09-14-module-decoupling-refactor-plan.md) - 权威停走门: [2026-09-14-document-reliability-m2-test-report.md](../../test-reports/2026-09-14-document-reliability-m2-test-report.md)（继续 M3） -- Test report: （本轮结束后回链） +- Test report: [2026-09-15-taskrun-runmanifest-u8-test-report.md](../../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) ## 证据 / 备注 diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 331672da..fd890979 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -5,7 +5,7 @@ ## 焦点 -1. **DOC-RELIABILITY**: [M3 U8 TaskRun/RunManifest](short-term/2026-09-15-taskrun-runmanifest-u8.md)（`feature/taskrun-runmanifest`；权威停走门 [M2 继续 M3](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；**未签字前禁止 U9–U12**） +1. **DOC-RELIABILITY**: [M3 U8 已签，继续 U9](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)（任务已归档 [2026-09-15-taskrun-runmanifest-u8](../archive/ops/2026-09-15-taskrun-runmanifest-u8.md)；U8 最后一句 **继续 U9**；**未签字前禁止 U10–U12**） 2. **CHAT-INTENT**: [对话意图运行时状态](../audit/2026-09-13-conversation-intent-runtime-state.md)（`fix/conversation-intent-state`；与文档账本运行时无关） 3. **CHAT**: [对话历史刷新](../audit/2026-09-13-chat-history-refresh-disappears.md)（认领仍 `in_progress`；gap 行已 PASS，以代码/handoff 为准收敛） @@ -13,7 +13,7 @@ | owner | task | branch | worktree | mutex | status | |-------|------|--------|----------|-------|--------| -| Cursor | taskrun-runmanifest-u8 | feature/taskrun-runmanifest | - | taskrun | in_progress | +| Cursor | taskrun-runmanifest-u8 | feature/taskrun-runmanifest | - | taskrun | done | | Cursor | document-reliability-m1-live-and-m2 | refactor/document-reliability-pilot | .worktrees/refactor-document-reliability-pilot | workspace | done | | Codex | module-decoupling-remediation-plan | feature/module-decoupling-remediation | .worktrees/feature/module-decoupling-remediation | architecture-docs | done | | verifier_antigravity | controlled-content-ui-acceptance | fix/controlled-content-ui-acceptance | .worktrees/fix-controlled-content-ui-acceptance | document-content-ui-acceptance | done | @@ -55,13 +55,13 @@ > 新任务开始前登记。`branch` 必须为 `feature/` 或 `fix/`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。 > workflow-plan 409 与治理鉴权 G1–G5 已收口（见 [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)）。禁止终止或改写项目 143。 -> `document-reliability-pilot` M2 已签（继续 M3）。U8 认领 `mutex=taskrun`。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未签字前禁止 U9–U12。 +> `document-reliability-pilot` U8 已签（继续 U9）。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未签字前禁止 U10–U12。 ## 集成槽位 | holder | phase | since | scope | queue | |--------|-------|-------|-------|-------| -| Cursor | merging | 2026-09-15 | M3 U8 feature/taskrun-runmanifest | - | +| - | idle | - | - | - | **phase**: `idle` | `merging` | `smoke` | `e2e` @@ -71,7 +71,7 @@ | 环境 | 端口 | schema | 状态 | |------|------|--------|------| -| dev | 8080 | **36** | healthy（`pi-agent`；M2 rebuild 后 health+ready PASS） | +| dev | 8080 | **37** | healthy（`pi-agent`；U8 rebuild 后 health+ready PASS） | | test | 8081 | 25→27* | healthy | | main | 8000 | 25→27* | healthy（app + runner + DB） | @@ -94,6 +94,7 @@ | 文档可靠性 M1 报告 | [../test-reports/2026-09-14-document-reliability-pilot-test-report.md](../test-reports/2026-09-14-document-reliability-pilot-test-report.md) | | 文档可靠性 M1 活验 | [../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md) | | 文档可靠性 M2 | [../test-reports/2026-09-14-document-reliability-m2-test-report.md](../test-reports/2026-09-14-document-reliability-m2-test-report.md) | +| 文档可靠性 M3 U8 | [../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | | SSE/幂等审计 | [../audit/2026-09-13-document-revision-sse-idempotency.md](../audit/2026-09-13-document-revision-sse-idempotency.md) | | 质量门禁自污染审计 | [../audit/2026-09-13-document-revision-quality-gate-self-pollution.md](../audit/2026-09-13-document-revision-quality-gate-self-pollution.md) | | 当前文档路由审计 | [../audit/2026-09-13-document-revision-active-doc-routing.md](../audit/2026-09-13-document-revision-active-doc-routing.md) | diff --git a/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md b/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md index f2a64590..bbb6df36 100644 --- a/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md +++ b/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md @@ -2,7 +2,7 @@ title: "模块解耦重构验收清单" status: in_progress created: 2026-09-14 -updated: 2026-09-14 +updated: 2026-09-15 owner: Cursor --- @@ -37,11 +37,11 @@ owner: Cursor ## M3 TaskRun、workspace 与数据生命周期 -- [ ] RunManifest 规范字节、SHA、成员和 storage identity 持久化。 -- [ ] 重复 finalize、SSE 断线和 cancel race 不重复消息或 artifact。 -- [ ] WorkspaceRef 无物理路径，路径遍历和链接逃逸负例 PASS。 -- [ ] 旧 lease writer 被 fencing token 拒绝。 -- [ ] workspace 清空后最终成果仍可定位和校验。 +- [x] RunManifest 规范字节、SHA、成员和 storage identity 持久化。 +- [x] 重复 finalize、SSE 断线和 cancel race 不重复消息或 artifact。 +- [x] WorkspaceRef 无物理路径，路径遍历和链接逃逸负例 PASS。 +- [x] 旧 lease writer 被 fencing token 拒绝。 +- [x] workspace 清空后最终成果仍可定位和校验。 - [ ] backfill 可暂停、续跑和重放；异常进入 quarantine。 - [ ] 八条 migration/reload/recovery 轨道 PASS。 - [ ] 数据销毁前完整备份、corpus、FAV coverage 和人工批准全部满足。 diff --git a/.doc/remediation/gap-matrix.md b/.doc/remediation/gap-matrix.md index f63c43eb..36fb99c5 100644 --- a/.doc/remediation/gap-matrix.md +++ b/.doc/remediation/gap-matrix.md @@ -70,7 +70,7 @@ | DOCREV-ACTIVE-DOC-ROUTING | UI-05「当前受控文档」被法规关键词改路由到 `feasibility_reg`，intake 无 pending | **PASS** | [audit](../audit/2026-09-13-document-revision-active-doc-routing.md)；Round20 UI-05 pending 停在 intake V2 | **不改弱 E2E**；不删证据 | | DOCREV-QUALITY-STATUS-FP | 行动计划表状态列「待续」被门禁当敷衍，两轮 repair 后 quality_failed 80 | **PASS** | [audit](../audit/2026-09-13-document-revision-quality-status-false-positive.md)；Round20 UI-03 | **不删禁词、不放宽正文**；不删证据 | | GOVERNANCE-ACTION-AUTHZ | 文档动作按职责鉴权；未激活 202-blocked；关闭创建直通 | **PASS** | [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)；[audit 异态](../audit/2026-09-14-planning-with-workflow-key-anomaly.md) | 不改 83 条 planning+key 夹具；none 长 PI 审批链另开 | -| DOC-RELIABILITY-PILOT | 受控文档写路径跨表非同事务、读路径被 projection 绑架、授权 TOCTOU；刷新/重启后可能 404/409/projection-missing | OPEN | [复核](../audit/2026-09-14-module-decoupling-plan-review.md)；[重构计划](./2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md) | M1 活验 **继续 M2**。M2 **继续 M3**。禁止在 M3 签字前写 U8–U12 业务代码 | +| DOC-RELIABILITY-PILOT | 受控文档写路径跨表非同事务、读路径被 projection 绑架、授权 TOCTOU；刷新/重启后可能 404/409/projection-missing | OPEN | [复核](../audit/2026-09-14-module-decoupling-plan-review.md)；[重构计划](./2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | M2 **继续 M3**。U8 **继续 U9**。禁止在 U9 签字前写 U10–U12 业务代码 | --- diff --git a/.doc/status/FEATURE-MATRIX.md b/.doc/status/FEATURE-MATRIX.md index 518e8bb3..8e40324b 100644 --- a/.doc/status/FEATURE-MATRIX.md +++ b/.doc/status/FEATURE-MATRIX.md @@ -1,6 +1,6 @@ # 模块与功能完成度（SSOT） -**更新**: 2026-09-14 +**更新**: 2026-09-15 **production_ready**: no **发布缺口**: [../remediation/gap-matrix.md](../remediation/gap-matrix.md) **协作焦点**: [../ops/CURRENT.md](../ops/CURRENT.md) @@ -27,7 +27,7 @@ | PI Runner / 配置链 | pi-agent + MiniMax 真实链路 | done | [test piagent](../test-reports/2026-09-13-codex-audit-p0-remediation-test-report.md)；[container-env](../test-reports/2026-09-13-container-env-configuration-test-report.md) | — | `pi-agent` | | 三环境运维 | dev/test/main + 共享 KB | done | [PROJECT.md](../../PROJECT.md) §4；四环境 healthy | — | `infra` | | CI / E2E | PG 集成 + 真实 PI E2E | partial | [gap-matrix BUG-08](../remediation/gap-matrix.md)；[test pg-ci](../test-reports/2026-09-12-pg-ci-soak-e2e-test-report.md)；[运行时 UI Round20](../test-reports/2026-09-14-document-revision-runtime-ui-test-report.md) | CI `TEST_DATABASE_URL` 隔离库仍缺；文档运行时真实 UI 已按 UI_PHASE 分文件且 post-restart 断言最终 A1 正文。`-race` 全包并行 TempDir flake 为基建残留 | `ci` | -| 模块解耦与可靠事件 | 文档可靠性纵向试点、跨属主事务、多副本互斥、FAV 与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[重构计划](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md) | U5–U7 PASS（WorkspaceRef/fencing、耐久投递、按文件 FAV）。活验 **继续 M2**；M2 最后一句 **继续 M3**。未开工 U8–U12 | `workspace` | +| 模块解耦与可靠事件 | 文档可靠性纵向试点、跨属主事务、多副本互斥、FAV 与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[重构计划](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | U5–U8 PASS。M2 **继续 M3**；U8 最后一句 **继续 U9**。未开工 U9–U12 | `taskrun` | --- diff --git a/.doc/test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md b/.doc/test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md new file mode 100644 index 00000000..ba03a2f8 --- /dev/null +++ b/.doc/test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md @@ -0,0 +1,124 @@ +# M3 U8 TaskRun 终态与 RunManifest 测试报告 + +## 一、 测试背景 (Background) + +权威停走门是 M2 报告最后一句 **继续 M3**（[2026-09-14-document-reliability-m2-test-report.md](./2026-09-14-document-reliability-m2-test-report.md)）。整改计划 U8 要求 TaskRun 终态与类型化 RunManifest 同一 UoW，并在重复 finalize、cancel 竞态、SSE 断线和 workspace 清理后保持一致。本轮不做 U9（dataset/八轨/销毁），不碰 M4–M6 / `chat` / `intent`。 + +## 二、 测试目标 (Test Objectives) + +- RunManifest 规范字节、SHA-256、workspace_id、storage_identity、members 持久化；Evidence/Gaps 不进核心协议。 +- 重复 finalize：同一 status、同一 manifest SHA、助手消息 1 条、envelope/staging 不加倍。 +- finalize ∥ cancel：恰好一个合法终态（双 pool，不用进程全局 mutex）。 +- SSE `done` 不出现在 TX 失败路径；commit 前失败后占位正文不落终态。 +- workspace 无 verified receipt 则 cleanup FAIL；有 receipt 则可清，且 staging 仍能取出原 SHA。 +- server 不再 `Join(workDir, "output.md")`；Backend 由服务端 policy 解析。 +- Goose 00037 Up+Down；占槽 rebuild 后 schema **37**。 + +## 三、 测试方案与环境 (Test Plan & Environment) + +| 项 | 值 | +|----|-----| +| OS | Linux 7.0.0-31-generic | +| Go | 本机 `go test`；容器 `agent-dev-app` | +| 分支 | `feature/taskrun-runmanifest`（从本地 `dev` 检出） | +| 部署 | 主仓 `./scripts/env-service.sh dev rebuild`（禁止 worktree 内 env-service） | +| Compose | 项目 `agent-dev`，HTTP `:8080` | +| Agent | `agent_backend=pi-agent` | +| Schema | **37**（00037 `task_run_manifests`） | +| PostgreSQL | 隔离库 `agent_rel_m3_u8`（admin 连 `shuimu-system-pg-test:5434/postgres` 自建）；测完 `DROP`；禁止 `agent_system_dev/test/main` | +| 项目 143 | **只读**；本轮未 terminate，未改 workflow-plan | +| 证据 | `.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/` | + +镜像身份：`app_image=sha256:fb3eec56ae5684fd323b1bd9e3c5c521c2e415730c2edd7f8c65a86b85aa6fb9`。见 `rebuild-health-ready.txt`。 + +提交：`2234af4` docs(ops) 认领 → `eb75f85` feat(db) → `7be8730` feat(taskrun) → `2ddaf76` feat(workspace) → `3997a04` test(taskrun)。 + +## 四、 预期结果 (Expected Results) + +| 栏 | 期望 | +|----|------| +| Unit | 规范 SHA 稳定；未授权 backend / 敏感正文拒绝；重复 finalize 不改 turn；cancel 后 finalize 采用已有终态 | +| PG | 双 pool 重复 finalize 计数均为 1；cancel race 唯一终态；commit 失败无 `done`；无 receipt 拒清理 | +| 故障注入 | `BeforeCommit` 失败回滚助手占位；`done` 不出现 | +| 真实 PI | 本轮未新开编制任务；dev 已 `pi-agent` ready；143 只读 | +| 浏览器 | web 单测 + production build 必做；页面回归继承 M1 | +| restart-rebuild | health/ready schema 37，`pi-agent` | +| 跨用户 | Backend 调用方不能塞任意字符串；敏感字段从 metadata 剥离 | + +## 五、 实际测试结果与判定 (Actual Results & Evaluation) + +### 5.1 Unit / stub + +- **测试输入**: `unset TEST_DATABASE_URL; go test -short ./internal/taskrun ./internal/workspace ./internal/server -run 'TestFinalize|TestCommands|TestRunManifest|TestCleanup|TestOpaque|TestManager_Done'` +- **实际输出**: `TestRunManifest_FreezeStableSHA`、`TestCommands_DuplicateFinalizeMemory`、`TestCommands_CancelWinsOverFinalizeMemory`、`TestCleanup_RequiresVerifiedReceipts`、`TestFinalizeTaskRun_HashesViaWorkspaceCapability`、`TestManager_FinalizeFailureDoesNotEmitDone` 均为 PASS。 +- **评价判定**: **PASS** +- **性能与耗时**: 亚秒。日志：`go-test-short.txt` + +### 5.2 PostgreSQL 隔离库（双 pool） + +- **测试输入**: `TEST_DATABASE_URL` 指向 5434 上非共享库名；`go test ./internal/taskrun -run TestPG_`。Admin 连 `postgres`，`CREATE DATABASE agent_rel_m3_u8`，goose 至 37。 +- **实际输出**: + - `TestPG_DuplicateFinalizeIdempotent`：messages=1 manifests=1 envelopes=1 staging=1，两侧 SHA 相同 + - `TestPG_FinalizeCancelRaceOneTerminal`：status ∈ {completed, cancelled}，助手 turn=1 + - `TestPG_CleanupWithoutReceiptFailsAndStagingSurvives`：无 receipt 目录仍在；有 receipt 后可删；`LoadStaging` SHA 与原文一致 +- **评价判定**: **PASS** +- **性能与耗时**: 含 goose 约 0.35s。测完 `DROP DATABASE`；`psql` 确认 `agent_rel_m3%` 为 0 行。 + +### 5.3 故障注入 / SSE + +- **测试输入**: `uow.PG.BeforeCommit` 返回 `inject commit fail`；Manager worker finalize。 +- **实际输出**: status=`finalizing_failed`；`task_run_events` 无 `done`；助手正文仍为「任务执行中…」。 +- **评价判定**: **PASS** +- **性能与耗时**: 40ms + +### 5.4 真实 PI + +- **测试输入**: 本轮未对新项目发起编制；`GET /health` 显示 `agent_backend=pi-agent`。项目 143 只读未打开。 +- **实际输出**: 模型容器在线，但 U8 未跑一条新的真实编制链路。 +- **评价判定**: **FAIL（未测）** — 不构成 U8 停走项（停走项是重复 finalize / 双终态 / 提前 `done` / 无 receipt 清理 / 仅进程 mutex）。浏览器页面回归继承 M1。 +- **性能与耗时**: 未测 + +### 5.5 浏览器 + +- **测试输入**: `cd web && npm test`；`npm run build` +- **实际输出**: tsx 365 passed + vitest 30 passed；production build ✓ 8.29s +- **评价判定**: **PASS**（单测+build）。页面 finalize 后刷新继承 M1，本轮未重跑 Playwright。 +- **性能与耗时**: 单测 ~1.3s；build 8.29s + +### 5.6 restart / rebuild + +- **测试输入**: 主仓占槽 `./scripts/env-service.sh dev rebuild`；`curl -sf http://127.0.0.1:8080/health` 与 `/ready` +- **实际输出**: + +```json +{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true} +``` + +- **评价判定**: **PASS** +- **性能与耗时**: rebuild 约 23s。日志：`rebuild.log`、`health.json`、`ready.json` + +### 5.7 跨用户 / Backend policy + +- **测试输入**: `TestResolveBackend_StripsUnauthorized`；`TestStripSecretsAndSensitive`；`TestCommands_RejectsSensitiveMarkdown` +- **实际输出**: `evil-shell` 拒绝；`api_key` 从 metadata 剥离；含 `api_key=` 的正文 finalize 失败 +- **评价判定**: **PASS** +- **性能与耗时**: 亚秒 + +## 六、 自动化测试套件汇总 (Test Suite Execution) + +| 命令 | 结果 | +|------|------| +| `go vet ./...` | PASS | +| `go test -short ./...` | PASS | +| 隔离库 goose 00037 Up + 双 pool 竞态 + `DROP DATABASE` | PASS | +| 目标包 `-race`（taskrun/workspace/ctxmgr） | PASS | +| `arch-check --mode warn` | PASS | +| web `npm test` | 365 + 30 PASS | +| web production build | PASS | +| 占槽 rebuild + health/ready | PASS，`schema_version=37` | + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +U8 停走项全部观测为 PASS：重复 finalize 不产生第二 turn / 第二份 staging；cancel 与 finalize 只有一个合法终态；TX 失败路径无 SSE `done`；无 verified receipt 不得删 workspace；正确性由 `SELECT … FOR UPDATE` 与 UNIQUE(run_id) 承担，而不是进程 mutex。Goose 00037 可随隔离库 Down（随 Down to 33 验证表消失）。server 已去掉 `filepath.Join(workDir, "output.md")`。未做 U9 backfill/corpus/八轨/销毁，未整支 merge FAV，未启用 `FAV_DEGRADED_READ`，未终止项目 143。真实 PI 编制本轮未新开，不作为 U8 扩面失败理由。 + +继续 U9 diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/arch-check-warn.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/arch-check-warn.txt new file mode 100644 index 00000000..f7cec0da --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/arch-check-warn.txt @@ -0,0 +1,4 @@ +arch-check mode=warn +boundaries=/home/weichen/projects/Agent-System/.doc/system-design/module-boundaries.yaml +ok chat-agent-system/internal/archtest 0.102s +arch-check warn: production gate passed diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-short.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-short.txt new file mode 100644 index 00000000..e11b4125 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-short.txt @@ -0,0 +1,47 @@ +ok chat-agent-system/cmd/bizdev-cli 2.342s +ok chat-agent-system/cmd/document-content-reconcile (cached) +? chat-agent-system/cmd/document-revision-backfill [no test files] +? chat-agent-system/cmd/recategorize [no test files] +? chat-agent-system/cmd/server [no test files] +ok chat-agent-system/internal/agent 1.178s +ok chat-agent-system/internal/app/docflow 0.003s +ok chat-agent-system/internal/app/uow 0.003s +ok chat-agent-system/internal/archtest 0.121s +ok chat-agent-system/internal/artifactvault 0.004s +ok chat-agent-system/internal/auth (cached) +ok chat-agent-system/internal/authz (cached) +ok chat-agent-system/internal/bizdev 0.161s +ok chat-agent-system/internal/chat 0.015s +ok chat-agent-system/internal/config (cached) +ok chat-agent-system/internal/coordinator 0.011s +ok chat-agent-system/internal/ctxmgr (cached) +ok chat-agent-system/internal/db 0.003s +ok chat-agent-system/internal/docduty (cached) +ok chat-agent-system/internal/docgov 0.018s +ok chat-agent-system/internal/docquery 0.003s +ok chat-agent-system/internal/docrevision 0.006s +ok chat-agent-system/internal/events 0.003s +ok chat-agent-system/internal/freedoc 0.027s +ok chat-agent-system/internal/fsutil (cached) +ok chat-agent-system/internal/intent 0.018s +ok chat-agent-system/internal/kb (cached) +ok chat-agent-system/internal/notification 0.070s +? chat-agent-system/internal/observability [no test files] +ok chat-agent-system/internal/projectcontext 0.004s +ok chat-agent-system/internal/projectinstance 0.004s +ok chat-agent-system/internal/projectworkflow (cached) +ok chat-agent-system/internal/quote (cached) +ok chat-agent-system/internal/replyexport (cached) +ok chat-agent-system/internal/server 10.015s +ok chat-agent-system/internal/squad 0.034s +ok chat-agent-system/internal/storage (cached) +ok chat-agent-system/internal/stream (cached) +ok chat-agent-system/internal/taskrun 0.146s +ok chat-agent-system/internal/template 0.709s +ok chat-agent-system/internal/upload (cached) +? chat-agent-system/internal/workflow [no test files] +ok chat-agent-system/internal/workflow/runtime (cached) +ok chat-agent-system/internal/workflowdef (cached) +ok chat-agent-system/internal/workspace (cached) +? chat-agent-system/templates [no test files] +SHORT_EXIT:0 diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-u8-packages-race.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-u8-packages-race.txt new file mode 100644 index 00000000..377f42e5 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-u8-packages-race.txt @@ -0,0 +1,4 @@ +ok chat-agent-system/internal/taskrun 1.961s +ok chat-agent-system/internal/workspace 1.012s +ok chat-agent-system/internal/ctxmgr 1.146s +RACE_EXIT:0 diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/health.json b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/health.json new file mode 100644 index 00000000..5720c76a --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/health.json @@ -0,0 +1 @@ +{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true} diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/ready.json b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/ready.json new file mode 100644 index 00000000..5720c76a --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/ready.json @@ -0,0 +1 @@ +{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true} diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/rebuild-health-ready.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/rebuild-health-ready.txt new file mode 100644 index 00000000..f7448815 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/rebuild-health-ready.txt @@ -0,0 +1,4 @@ +sha256:fb3eec56ae5684fd323b1bd9e3c5c521c2e415730c2edd7f8c65a86b85aa6fb9 [agent-dev-app@sha256:fb3eec56ae5684fd323b1bd9e3c5c521c2e415730c2edd7f8c65a86b85aa6fb9] +{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true} + +{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true}