diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 4d22423f..4881edf5 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -6,13 +6,14 @@ ## 焦点 1. **CHAT-INTENT**: [对话意图运行时状态](../audit/2026-09-13-conversation-intent-runtime-state.md)（`fix/conversation-intent-state` 已是 `dev` 祖先；浏览器双标签 busy 仍待 Codex 收口） -2. **DOC-RELIABILITY**: [U10 已合入，停止扩面](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md)（占槽 rebuild PASS；**禁止 U11–U12**） +2. **DOC-RELIABILITY**: [U11–U12 组合根与传输门禁](./short-term/2026-09-15-app-composition-u11-u12.md)（用户已授权；合同 PASS，待占槽合入 `dev`） 3. **CHAT**: [对话历史刷新活验已补](../test-reports/2026-09-13-chat-history-refresh-test-report.md)（项目 6252 闲聊刷新同集 + guest 403；编制任务链仍开放） ## 并行认领 | owner | task | branch | worktree | mutex | status | |-------|------|--------|----------|-------|--------| +| Cursor | app-composition-u11-u12 | feature/app-composition-u11-u12 | - | app | in_progress | | Cursor | chatlog-contextasm-u10 | feature/chatlog-contextasm-u10 | - | chat | done | | Cursor | chat-history-refresh-remediation | fix/chat-history-refresh | - | chat | done | | Cursor | dataset-lifecycle-u9 | feature/dataset-lifecycle-u9 | - | dataset | done | @@ -57,13 +58,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` U10 **停止扩面**（已合入 `dev`，占槽 rebuild PASS）。禁止 U11–U12。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。不碰 `intent`。 +> `document-reliability-pilot` U11–U12 合同已签（`feature/app-composition-u11-u12`）。合入 `dev` 须占集成槽。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。不碰 `intent`。 ## 集成槽位 | holder | phase | since | scope | queue | |--------|-------|-------|-------|-------| -| - | idle | - | - | - | +| Cursor | merging | 2026-09-15 | U11-U12 app-composition | intent dual-tab after idle | **phase**: `idle` | `merging` | `smoke` | `e2e` @@ -99,6 +100,7 @@ | 文档可靠性 M3 U8 | [../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | | 文档可靠性 M3 U9 | [../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md) | | 文档可靠性 M4 U10 | [../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md) | +| 文档可靠性 M5/M6 U11–U12 | [../test-reports/2026-09-15-app-composition-u11-u12-test-report.md](../test-reports/2026-09-15-app-composition-u11-u12-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 f267b95f..ad72905e 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 @@ -50,10 +50,10 @@ owner: Cursor - [x] 每个子切片在原包边界先获得行为收益，再决定物理搬包。 - [x] chatlog/contextasm 保持授权可见集合和计数一致。 -- [ ] app 是唯一组合根，subscriber 重复启动不重复注册。 -- [ ] UserProjectContext 不再被多个模块字段级直改。 -- [ ] server 无 SQL、路径拼接和业务闭包。 -- [ ] 对应架构规则全部 block，过期例外和 flags 已删除。 +- [x] app 是唯一组合根，subscriber 重复启动不重复注册。 +- [x] UserProjectContext 跨模块写路径走类型化 ContextUpdate + 版本 CAS（coordinator 作为 owner 仍可改 Project 字段）。 +- [ ] server 无 SQL、路径拼接和业务闭包。（OnOutbox 已迁出；生产 SQL 无新增；path 例外 `workspaces/` / `output.md` 未到期，未删） +- [ ] 对应架构规则全部 block，过期例外和 flags 已删除。（`import`/`event` 已 block；sql_write/path 基线例外仍在） ## 证据分层 @@ -69,6 +69,6 @@ owner: Cursor ## 关闭条件 -- [ ] test report 回链具体 U-ID、commit、环境 identity 和 fixture/dataset。（U10 已回链；U11–U12 未实施，禁止用 U10 冒充全计划完成） +- [x] test report 回链具体 U-ID、commit、环境 identity 和 fixture/dataset。（[U11–U12 报告](../../test-reports/2026-09-15-app-composition-u11-u12-test-report.md)；合入 `dev` 后补 rebuild 镜像） - [x] CURRENT、FEATURE-MATRIX、gap-matrix 与任务状态一致。 - [x] 未实施后续仍为 planned/deferred，不用局部测试冒充全计划完成。 diff --git a/.doc/ops/short-term/2026-09-15-app-composition-u11-u12.md b/.doc/ops/short-term/2026-09-15-app-composition-u11-u12.md new file mode 100644 index 00000000..43ad8446 --- /dev/null +++ b/.doc/ops/short-term/2026-09-15-app-composition-u11-u12.md @@ -0,0 +1,61 @@ +--- +title: "M5/M6 组合根与传输门禁 U11–U12" +status: in_progress +created: 2026-09-15 +updated: 2026-09-15 +owner: Cursor +branch: feature/app-composition-u11-u12 +worktree: "-" +mutex: app +--- + +# M5/M6 组合根与传输门禁 U11–U12 + +## 目标 + +按整改计划四个子切片完成组合根、文档通知闭包迁出、stageflow 用例、UserProjectContext CAS；再收口 SSE 编码与已清零规则的 arch-check block。用户已明确授权继续 U11/U12（推翻 U10 报告的停止扩面）。 + +## 范围 + +- U11a：`internal/app` 唯一 composition root；幂等 Start/Stop +- U11b：文档 `OnOutbox` 通知迁到 `internal/app/docnotify` +- U11c：`internal/app/stageflow` 拥有 ExecuteTask / ExecuteStage +- U11d：`ContextUpdate` 版本 CAS，禁止跨模块字段级直改 +- U12：SSE 中性 frame 编码；`import`/`event` 转 block；不改错误信封与 PI Runner 协议 + +## 非目标 + +- 不碰 `intent` 包行为（mutex 仍由 Codex 占用） +- 不整支 merge `feature/final-artifact-vault`；不启用 `FAV_DEGRADED_READ` +- 不终止或改写项目 143 +- 不删除未到期的 sql_write/path 例外（U3/U5 残留） +- 不在 worktree 内 `env-service` + +## 步骤 + +1. [x] 认领 CURRENT + 本任务 +2. [x] U11a composition root +3. [x] U11b docnotify +4. [x] U11c stageflow +5. [x] U11d ContextUpdate +6. [x] U12 SSE + arch-check block_kinds +7. [x] 短测 / 目标包 race / 报告 + +## 验收标准 + +- [x] 重复 Start 不重复注册 subscriber；Stop 后再 Start worker 仍在 +- [x] server 生产代码不再赋值 `OnOutbox` +- [x] HTTP 与 CLI 走同一 stageflow 用例 +- [x] ContextUpdate 并发 stale version 不丢更新赢家 +- [x] `arch-check --mode block` 对已清零的 import/event 通过 +- [x] 测试报告回链 U11/U12 + +## 相关链接 + +- Checklist: [../checklists/2026-09-14-module-decoupling-refactor-acceptance.md](../checklists/2026-09-14-module-decoupling-refactor-acceptance.md) +- Plan: [../../remediation/2026-09-14-module-decoupling-refactor-plan.md](../../remediation/2026-09-14-module-decoupling-refactor-plan.md) +- Test report: [../../../test-reports/2026-09-15-app-composition-u11-u12-test-report.md](../../../test-reports/2026-09-15-app-composition-u11-u12-test-report.md) + +## 证据 / 备注 + +用户口头授权：「继续解耦到 U11、U12」。 diff --git a/.doc/remediation/gap-matrix.md b/.doc/remediation/gap-matrix.md index 2afaf71f..0b689f60 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)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)；[M3 U9](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md)；[U10 停止扩面](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md) | M3 已签。U10 **停止扩面**（未拆包；已合入 `dev` + rebuild）。禁止 U11–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)；[M3 U9](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md)；[U10 报告](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md)；[U11–U12 报告](../test-reports/2026-09-15-app-composition-u11-u12-test-report.md) | U11–U12 合同已签，待占槽合入 `dev` + rebuild。path/sql_write 未到期例外保留。不碰 intent / 项目 143 | --- diff --git a/.doc/status/FEATURE-MATRIX.md b/.doc/status/FEATURE-MATRIX.md index 2c5de0f0..9e5f62fd 100644 --- a/.doc/status/FEATURE-MATRIX.md +++ b/.doc/status/FEATURE-MATRIX.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)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)；[M3 U9](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md)；[U10 报告](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md) | U5–U10 已签（U10 **停止扩面**，已合入 `dev` + rebuild）。禁止 U11–U12。未整支 merge FAV | `architecture-docs` | +| 模块解耦与可靠事件 | 文档可靠性纵向试点、跨属主事务、多副本互斥、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)；[M3 U9](../test-reports/2026-09-15-dataset-lifecycle-u9-test-report.md)；[U10 报告](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md)；[U11–U12 报告](../test-reports/2026-09-15-app-composition-u11-u12-test-report.md) | U5–U12 合同已签（U11–U12 在 `feature/app-composition-u11-u12`，合入 `dev` 后占槽 rebuild）。path/sql_write 未到期例外保留。未整支 merge FAV | `architecture-docs` | --- diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 14748570..7540b0ca 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -56,12 +56,36 @@ imports: - storage db: [] docduty: [] - app: [] + app: + - agent + - app/chatflow + - app/docnotify + - app/stageflow + - app/uow + - artifactvault + - auth + - authz + - bizdev + - config + - coordinator + - docgov + - intent + - notification + - observability + - projectinstance + - projectworkflow + - taskrun + - workflowdef + - workspace app/chatflow: - bizdev - coordinator - ctxmgr - projectinstance + app/docnotify: + - docgov + - docrevision + - notification app/docflow: - app/uow - docgov @@ -69,6 +93,8 @@ imports: - docrevision - events app/uow: [] + app/stageflow: + - bizdev docquery: - docgov - docrevision @@ -105,8 +131,10 @@ imports: replyexport: [] server: - agent + - app - app/chatflow - app/docflow + - app/stageflow - app/uow - auth - authz @@ -128,6 +156,7 @@ imports: - replyexport - squad - stream + - stream/transport - taskrun - upload - workflowdef @@ -142,6 +171,8 @@ imports: - workspace storage: [] stream: [] + stream/transport: + - stream taskrun: - app/uow - artifactvault @@ -256,7 +287,7 @@ event_producers: event_consumers: outbox.notify: - - server + - app/docnotify - notification unwrap_owners: @@ -443,3 +474,9 @@ baseline_counts: sql_write: 5 path: 16 event: 0 + +# U12: kinds already at zero production findings fail --mode block. +# sql_write/path keep dated exceptions until their owner milestones clear them. +block_kinds: + - import + - event diff --git a/.doc/test-reports/2026-09-15-app-composition-u11-u12-test-report.md b/.doc/test-reports/2026-09-15-app-composition-u11-u12-test-report.md new file mode 100644 index 00000000..b81798ca --- /dev/null +++ b/.doc/test-reports/2026-09-15-app-composition-u11-u12-test-report.md @@ -0,0 +1,101 @@ +# M5/M6 U11–U12 组合根与传输门禁 测试报告 + +## 一、 测试背景 (Background) + +权威停走门是 U10 报告最后一句 **停止扩面**（[2026-09-15-chatlog-contextasm-u10-test-report.md](./2026-09-15-chatlog-contextasm-u10-test-report.md)）。用户于 2026-09-15 明确授权「继续解耦到 U11、U12」，推翻该停止扩面。本轮按整改计划四个子切片落地组合根、文档通知闭包迁出、stageflow 用例、UserProjectContext CAS，以及 SSE 中性编码与已清零规则的 arch-check block。 + +## 二、 测试目标 (Test Objectives) + +- U11a：`internal/app.Runtime` 是唯一 composition root；重复 Start 不重复注册 subscriber；Stop 后再 Start worker 仍在。 +- U11b：文档 `OnOutbox` 从 `internal/server` 迁到 `internal/app/docnotify`；Bind 幂等。 +- U11c：HTTP / CLI / coordinator 走同一 `stageflow.UseCase`。 +- U11d：`ApplyContextUpdate` 版本 CAS，并发同版本恰好一个赢家。 +- U12：SSE 只编码中性 frame；`import`/`event` 进入 `--mode block`；不改错误信封与 PI Runner 协议。 +- 不碰 `intent`；不终止项目 143；不启用 `FAV_DEGRADED_READ`。 + +## 三、 测试方案与环境 (Test Plan & Environment) + +| 项 | 值 | +|----|-----| +| OS | Linux 7.0.0-31-generic | +| 分支 | `feature/app-composition-u11-u12`（从本地 `dev@ef9a162` 检出） | +| 本机合同测试 | `unset TEST_DATABASE_URL; go test -short ./internal/app ./internal/app/docnotify ./internal/app/stageflow ./internal/coordinator ./internal/stream/transport ./internal/archtest` | +| 架构门 | `./scripts/arch-check.sh --mode warn` 与 `--mode block` | +| 目标包 race | `go test -race -short` 上述包（不含全仓 race） | +| 全仓 short | `unset TEST_DATABASE_URL; go test -short -count=1 ./...` | +| 真实 PI / 浏览器 / rebuild | 本分支未占集成槽；合入 `dev` 后由主仓 rebuild 证明镜像 | +| 证据 | `.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/` | + +## 四、 预期结果 (Expected Results) + +| 栏 | 期望 | +|----|------| +| Start 幂等 | 连续两次 Start，`StartCount==1` 且 `SubscriberCount==1` | +| Restart | Stop 后再 Start，`StartCount==2` 且 subscriber 不加倍 | +| OnOutbox | 生产 `internal/server` 无 `OnOutbox =`；`docnotify.Bind` 第二次为 no-op | +| stageflow | HTTP 与 CLI 均调用 `stageflow.UseCase`，同一 stub 计数共享 | +| CAS | 两 goroutine 同 `ExpectedVersion` → 1 成功 / 1 `ErrStaleContext` | +| SSE | frame 含 `event:`/`data:`，不含 `retry:` | +| arch-check block | `import`/`event` 计数为 0；sql_write/path 既有例外仍允许 | + +## 五、 实际测试结果与判定 (Actual Results & Evaluation) + +### 5.1 U11a composition root + +- **测试输入**: `go test -short -count=1 ./internal/app -run TestStartStopRestartIdempotent -v` +- **实际输出**: `TestStartStopRestartIdempotent` PASS；`cmd/server` 改为 `app.New` + `Runtime.Start` + `server.NewFromRuntime` +- **评价判定**: **PASS** +- **性能与耗时**: 0.014s。日志：`assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages.txt` + +### 5.2 U11b 文档通知闭包 + +- **测试输入**: `go test -short -count=1 ./internal/app/docnotify -v`；`rg OnOutbox internal/server` +- **实际输出**: `TestBindOnce` / `TestHandleCandidateReady` / `TestHandleSkipsUnknownAndNoRecipient` PASS。生产 server 包无 `OnOutbox`。`event_consumers.outbox.notify` 属主改为 `app/docnotify` +- **评价判定**: **PASS** +- **性能与耗时**: 0.002s + +### 5.3 U11c stageflow + +- **测试输入**: `go test -short -count=1 ./internal/app/stageflow -v` +- **实际输出**: `TestUseCaseDelegates` PASS（HTTP/CLI 两个 UseCase 共享同一 stub，task=1 stage=1）；`handlers` / `task_run_handlers` / `cmd/bizdev-cli` / coordinator `SetStageTaskRunner` 均走该用例 +- **评价判定**: **PASS** +- **性能与耗时**: 0.004s + +### 5.4 U11d ContextUpdate CAS + +- **测试输入**: `go test -race -short -count=1 ./internal/coordinator -run TestContextUpdateConcurrentNoLostUpdate -v` +- **实际输出**: 两 goroutine 同版本写 blackboard：1 赢家、1 `ErrStaleContext`，`Version==2`。`-race` PASS（零值 `sync.Mutex`，无懒初始化竞态） +- **评价判定**: **PASS** +- **性能与耗时**: 约 1.0s（race）。日志：`assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages-race.txt` + +说明：`UserProjectContext` 字段仍导出供读取。跨模块写路径（server 会话组装、chatflow assembler、chat session、coordinator pending/card/blackboard）已改 `ApplyContextUpdate`。coordinator 内部仍有 `Project.CurrentStage` 等字段赋值（该包为上下文 owner）。 + +### 5.5 U12 SSE 与 arch-check + +- **测试输入**: `go test -short ./internal/stream/transport`；`./scripts/arch-check.sh --mode warn`；`./scripts/arch-check.sh --mode block` +- **实际输出**: `TestWriteNeutralFrame` / `TestHeartbeat` PASS。warn 与 block 均为 `production gate passed`。CI 增加 `arch-check --mode block`。`block_kinds: [import, event]`。未到期 sql_write(5)/path(16) 例外保留 +- **评价判定**: **PASS**（已清零规则 block；路径/SQL writer 基线例外未假装清零） +- **性能与耗时**: arch-check block 约 0.2s。日志：`arch-check-warn.txt`、`arch-check-block.txt` + +### 5.6 真实 PI / 浏览器 / rebuild + +- 真实 PI：本切片不改生成链路，未重跑编制任务。 +- 浏览器：未改 web 交互。 +- rebuild：未占集成槽；合入 `dev` 后须主仓 `./scripts/env-service.sh dev rebuild` 证明镜像。当前 `:8080` 仍为 U10 镜像。 + +## 六、 自动化测试套件汇总 (Test Suite Execution) + +| 命令 | 结果 | +|------|------| +| `go vet ./...` | PASS | +| `unset TEST_DATABASE_URL; go test -short -count=1 ./internal/app ./internal/app/docnotify ./internal/app/stageflow ./internal/coordinator ./internal/stream/transport ./internal/archtest` | PASS | +| `go test -race -short -count=1` 同上目标包（不含 archtest） | PASS | +| `./scripts/arch-check.sh --mode warn` | PASS | +| `./scripts/arch-check.sh --mode block` | PASS（import/event 已清零） | +| `unset TEST_DATABASE_URL; go test -short -count=1 ./...` | PASS（EXIT:0，`internal/server` 9.992s） | + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +U11a–U11d 与 U12 合同测试达到合入标准。未清零的 server `workspaces/` / `output.md` 路径例外与 sql_write 基线仍按到期日保留，不把 `--mode block` 写成全规则清零。合入 `dev` 需占 CURRENT 集成槽并 rebuild。 + +**停止扩面**（整改计划 U1–U12 已在本分支落地；未到期 path/sql_write 例外不属于本单元删除范围）。 diff --git a/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/arch-check-block.txt b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/arch-check-block.txt new file mode 100644 index 00000000..2ae3f1b2 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/arch-check-block.txt @@ -0,0 +1,5 @@ +arch-check mode=block +boundaries=/home/weichen/projects/Agent-System/.doc/system-design/module-boundaries.yaml +ok chat-agent-system/internal/archtest 0.110s +ok chat-agent-system/internal/archtest 0.105s +arch-check block: production gate passed diff --git a/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/arch-check-warn.txt b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/arch-check-warn.txt new file mode 100644 index 00000000..e8cb4ec1 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/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.101s +arch-check warn: production gate passed diff --git a/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-short.txt b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-short.txt new file mode 100644 index 00000000..81ee932c --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-short.txt @@ -0,0 +1,53 @@ +ok chat-agent-system/cmd/bizdev-cli 2.383s +? chat-agent-system/cmd/dataset [no test files] +ok chat-agent-system/cmd/document-content-reconcile 0.005s +? 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.076s +ok chat-agent-system/internal/app 0.014s +ok chat-agent-system/internal/app/chatflow 0.007s +ok chat-agent-system/internal/app/docflow 0.005s +ok chat-agent-system/internal/app/docnotify 0.002s +ok chat-agent-system/internal/app/stageflow 0.005s +ok chat-agent-system/internal/app/uow 0.004s +ok chat-agent-system/internal/archtest 0.138s +ok chat-agent-system/internal/artifactvault 0.005s +ok chat-agent-system/internal/auth 0.838s +ok chat-agent-system/internal/authz 0.003s +ok chat-agent-system/internal/bizdev 0.164s +ok chat-agent-system/internal/chat 0.016s +ok chat-agent-system/internal/config 0.012s +ok chat-agent-system/internal/coordinator 0.013s +ok chat-agent-system/internal/ctxmgr 0.097s +ok chat-agent-system/internal/dataset 0.012s +ok chat-agent-system/internal/db 0.012s +ok chat-agent-system/internal/docduty 0.012s +ok chat-agent-system/internal/docgov 0.021s +ok chat-agent-system/internal/docquery 0.012s +ok chat-agent-system/internal/docrevision 0.012s +ok chat-agent-system/internal/events 0.012s +ok chat-agent-system/internal/freedoc 0.027s +ok chat-agent-system/internal/fsutil 0.013s +ok chat-agent-system/internal/intent 0.018s +ok chat-agent-system/internal/kb 0.012s +ok chat-agent-system/internal/notification 0.078s +? chat-agent-system/internal/observability [no test files] +ok chat-agent-system/internal/projectcontext 0.015s +ok chat-agent-system/internal/projectinstance 0.016s +ok chat-agent-system/internal/projectworkflow 0.014s +ok chat-agent-system/internal/quote 0.019s +ok chat-agent-system/internal/replyexport 0.013s +ok chat-agent-system/internal/server 9.992s +ok chat-agent-system/internal/squad 0.039s +ok chat-agent-system/internal/storage 0.017s +ok chat-agent-system/internal/stream 0.003s +ok chat-agent-system/internal/stream/transport 0.003s +ok chat-agent-system/internal/taskrun 0.146s +ok chat-agent-system/internal/template 0.722s +ok chat-agent-system/internal/upload 0.003s +? chat-agent-system/internal/workflow [no test files] +ok chat-agent-system/internal/workflow/runtime 0.003s +ok chat-agent-system/internal/workflowdef 0.003s +ok chat-agent-system/internal/workspace 0.003s +? chat-agent-system/templates [no test files] diff --git a/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages-race.txt b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages-race.txt new file mode 100644 index 00000000..ba277d9e --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages-race.txt @@ -0,0 +1,5 @@ +ok chat-agent-system/internal/app 1.025s +ok chat-agent-system/internal/app/docnotify 1.008s +ok chat-agent-system/internal/app/stageflow 1.015s +ok chat-agent-system/internal/coordinator 1.041s +ok chat-agent-system/internal/stream/transport 1.008s diff --git a/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages.txt b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages.txt new file mode 100644 index 00000000..2fdfd373 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-test-u11-packages.txt @@ -0,0 +1,6 @@ +ok chat-agent-system/internal/app 0.014s +ok chat-agent-system/internal/app/docnotify 0.002s +ok chat-agent-system/internal/app/stageflow 0.004s +ok chat-agent-system/internal/coordinator 0.007s +ok chat-agent-system/internal/stream/transport 0.002s +ok chat-agent-system/internal/archtest 0.102s diff --git a/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-vet.txt b/.doc/test-reports/assets/2026-09-15-app-composition-u11-u12/go-vet.txt new file mode 100644 index 00000000..e69de29b diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 432cd14a..feb91fe5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -45,6 +45,9 @@ jobs: TEST_DATABASE_URL: postgres://app:ci_secret_db_pass@localhost:5432/agent_system_ci?sslmode=disable run: go test -short -timeout 20m ./... + - name: arch-check (block cleared kinds) + run: ./scripts/arch-check.sh --mode block + backup-restore-drill: runs-on: ubuntu-latest services: diff --git a/cmd/bizdev-cli/main.go b/cmd/bizdev-cli/main.go index 1b87b878..f40a8fc1 100644 --- a/cmd/bizdev-cli/main.go +++ b/cmd/bizdev-cli/main.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "chat-agent-system/internal/app/stageflow" "chat-agent-system/internal/bizdev" "chat-agent-system/internal/chat" "chat-agent-system/internal/config" @@ -240,7 +241,7 @@ func main() { return } - res, err := svc.ExecuteTask(context.Background(), bizdev.TaskRunRequest{ + res, err := stageflow.New(svc).ExecuteTask(context.Background(), bizdev.TaskRunRequest{ TaskKey: *taskKey, ProjectName: *project, CustomerName: *customer, @@ -347,7 +348,7 @@ func handleStageRun( ctx, _ = stream.WithListener(ctx, verboseMode, listener) } - stageRes, err := svc.ExecuteStage(ctx, bizdev.StageRunRequest{ + stageRes, err := stageflow.New(svc).ExecuteStage(ctx, bizdev.StageRunRequest{ Stage: bizdev.MaturityStage(stageName), ProjectName: project, CustomerName: customer, diff --git a/cmd/server/main.go b/cmd/server/main.go index 39e2a761..71b260ab 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -10,7 +10,7 @@ import ( "syscall" "time" - "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/app" "chat-agent-system/internal/config" "chat-agent-system/internal/server" ) @@ -38,12 +38,16 @@ func main() { cfg.AgentBackend = *backend } - svc, err := bizdev.NewService(cfg) + rt, err := app.New(cfg) if err != nil { fmt.Fprintf(os.Stderr, "业务服务初始化失败: %v\n", err) os.Exit(1) } - srv := server.NewServer(cfg, svc) + if err := rt.Start(context.Background()); err != nil { + fmt.Fprintf(os.Stderr, "后台 worker 启动失败: %v\n", err) + os.Exit(1) + } + srv := server.NewFromRuntime(rt) addr := fmt.Sprintf("%s:%d", cfg.HTTPHost, cfg.HTTPPort) fmt.Println("==================================================================") diff --git a/internal/app/app.go b/internal/app/app.go new file mode 100644 index 00000000..1b400912 --- /dev/null +++ b/internal/app/app.go @@ -0,0 +1,305 @@ +package app + +import ( + "context" + "fmt" + "log/slog" + "sync" + "time" + + "chat-agent-system/internal/agent" + "chat-agent-system/internal/app/chatflow" + "chat-agent-system/internal/app/docnotify" + "chat-agent-system/internal/app/stageflow" + "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" + "chat-agent-system/internal/config" + "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/intent" + "chat-agent-system/internal/notification" + "chat-agent-system/internal/observability" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/taskrun" + "chat-agent-system/internal/workflowdef" + "chat-agent-system/internal/workspace" +) + +const defaultReminderPollInterval = 15 * time.Minute + +// Runtime is the unique composition root. cmd/server, tests and workers +// construct it once; HTTP adapters must not re-wire domain services. +type Runtime struct { + Cfg *config.Config + Biz *bizdev.Service + Engine *coordinator.CoordinatorEngine + Auth *auth.Service + PW *projectworkflow.Service + TaskRuns *taskrun.Manager + Authorizer *authz.Authorizer + Authz *authz.Service + WFDef *workflowdef.Service + Inst *projectinstance.Service + Notify *notification.Service + Resolver *workspace.Resolver + WSCap *workspace.Capability + Commands *taskrun.Commands + Rec intent.Recognizer + Stageflow *stageflow.UseCase + DocNotify *docnotify.Subscriber + ChatLog chatflow.ChatLog + Assembler chatflow.ContextAssembler + + mu sync.Mutex + running bool + subscribers int + starts int + notifyBound bool + hookBound bool + reminderCancel context.CancelFunc +} + +func New(cfg *config.Config) (*Runtime, error) { + return Assemble(cfg, nil) +} + +// Assemble wires domain services. svc may be injected by tests. +func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { + if cfg == nil { + cfg, _ = config.Load("") + } + if cfg != nil { + observability.InitLogger(cfg.Env) + } + if svc == nil { + var err error + svc, err = bizdev.NewService(cfg) + if err != nil { + return nil, err + } + } + + rec := intent.NewRecognizer() + dialogueBackend := agent.NewBackend(cfg.AgentBackend, cfg) + coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir) + pool := svc.Pool() + stores, storeErr := auth.NewStores(context.Background(), pool, cfg) + if storeErr != nil { + return nil, fmt.Errorf("auth store: %w", storeErr) + } + authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg) + pwSvc := svc.ProjectWorkflow() + if pwSvc == nil { + pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil) + } + + authzSvc := authz.NewService(pool) + wfDefSvc := workflowdef.NewService(pool, svc.CAS()) + instStore := projectinstance.NewStore(pool) + var instSvc *projectinstance.Service + if wfDefSvc.Available() { + docChecker := projectinstance.NewDocGovChecker(svc.DocGov()) + instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker) + if pool != nil { + instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool)) + } + } + taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager()) + 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) + + stages := stageflow.New(svc) + coordEngine.SetStageTaskRunner(stages) + if instSvc != nil && instSvc.Available() { + coordEngine.SetInstanceService(instSvc) + } + + rt := &Runtime{ + Cfg: cfg, + Biz: svc, + Engine: coordEngine, + Auth: authSvc, + PW: pwSvc, + TaskRuns: taskMgr, + Authorizer: authz.NewAuthorizer(pool), + Authz: authzSvc, + WFDef: wfDefSvc, + Inst: instSvc, + Notify: notification.NewService(pool), + Resolver: workspace.NewResolver(cfg.DataDir, pool), + WSCap: wsCap, + Commands: cmds, + Rec: rec, + Stageflow: stages, + } + rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: svc.DocGov()} + rt.bindDocumentNotify() + rt.bindWorkflowApprovalHook() + return rt, nil +} + +func (r *Runtime) bindDocumentNotify() { + if r == nil || r.notifyBound { + return + } + if r.Biz != nil && r.Biz.DocRevision() != nil { + if docnotify.Bind(r.Biz.DocRevision(), r.DocNotify) { + r.subscribers++ + } + } + r.notifyBound = true +} + +func (r *Runtime) bindWorkflowApprovalHook() { + if r == nil || r.hookBound || r.Biz == nil || r.Biz.DocGov() == nil || r.Inst == nil || !r.Inst.Available() { + r.hookBound = true + return + } + r.Biz.DocGov().AddOnApprovedHook(r.onWorkflowDocumentApproved) + r.hookBound = true + r.subscribers++ +} + +func (r *Runtime) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) { + if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || r.Inst == nil { + return + } + result, err := r.Inst.HandleDocumentApproved(ctx, doc.ProjectID, doc.TaskKey) + if err != nil || result == nil || !result.StageReady { + return + } + if r.Notify == nil || !r.Notify.Available() { + return + } + if r.PW == nil { + return + } + project, err := r.PW.GetProject(ctx, doc.ProjectID, 0) + if err != nil || project == nil || project.CreatorUserID <= 0 { + return + } + _, _ = r.Notify.EmitStageReadyToAdvance(ctx, project.CreatorUserID, notification.StageReadyPayload{ + ProjectID: doc.ProjectID, + WorkflowInstanceID: result.InstanceID, + StageInstanceID: result.StageInstanceID, + StageKey: result.StageKey, + StageName: result.StageName, + ConfigRevision: result.ConfigRevision, + RequiredDocCount: result.RequiredDocCount, + ApprovedDocCount: result.ApprovedDocCount, + }) +} + +// Start launches background workers. Repeat calls on a running runtime are no-ops. +func (r *Runtime) Start(_ context.Context) error { + if r == nil { + return fmt.Errorf("app: nil runtime") + } + r.mu.Lock() + defer r.mu.Unlock() + if r.running { + return nil + } + r.bindDocumentNotify() + r.bindWorkflowApprovalHook() + ctx, cancel := context.WithCancel(context.Background()) + r.reminderCancel = cancel + go r.runReminderLoop(ctx) + r.running = true + r.starts++ + if r.subscribers == 0 { + r.subscribers = 1 + } + return nil +} + +// Stop cancels workers. A later Start recreates the reminder loop without +// double-registering document subscribers. +func (r *Runtime) Stop(_ context.Context) error { + if r == nil { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + if !r.running { + return nil + } + if r.reminderCancel != nil { + r.reminderCancel() + r.reminderCancel = nil + } + r.running = false + return nil +} + +func (r *Runtime) SubscriberCount() int { + if r == nil { + return 0 + } + r.mu.Lock() + defer r.mu.Unlock() + return r.subscribers +} + +func (r *Runtime) StartCount() int { + if r == nil { + return 0 + } + r.mu.Lock() + defer r.mu.Unlock() + return r.starts +} + +func (r *Runtime) Running() bool { + if r == nil { + return false + } + r.mu.Lock() + defer r.mu.Unlock() + return r.running +} + +func (r *Runtime) runReminderLoop(ctx context.Context) { + r.runStageAdvanceReminders(ctx) + ticker := time.NewTicker(defaultReminderPollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.runStageAdvanceReminders(ctx) + } + } +} + +func (r *Runtime) runStageAdvanceReminders(ctx context.Context) { + if r.Notify == nil || !r.Notify.Available() { + return + } + n, err := r.Notify.ProcessStageAdvanceReminders(ctx) + if err != nil { + slog.Warn("stage advance reminder scan failed", "error", err) + return + } + if n > 0 { + slog.Info("stage advance 24h reminders sent", "count", n) + } +} diff --git a/internal/app/app_test.go b/internal/app/app_test.go new file mode 100644 index 00000000..e3c281f6 --- /dev/null +++ b/internal/app/app_test.go @@ -0,0 +1,44 @@ +package app + +import ( + "context" + "testing" + "time" +) + +func TestStartStopRestartIdempotent(t *testing.T) { + rt := &Runtime{} + ctx := context.Background() + if err := rt.Start(ctx); err != nil { + t.Fatal(err) + } + if err := rt.Start(ctx); err != nil { + t.Fatal(err) + } + if rt.StartCount() != 1 { + t.Fatalf("duplicate Start spawned workers: starts=%d", rt.StartCount()) + } + if got := rt.SubscriberCount(); got != 1 { + t.Fatalf("subscribers=%d", got) + } + if !rt.Running() { + t.Fatal("expected running") + } + if err := rt.Stop(ctx); err != nil { + t.Fatal(err) + } + if rt.Running() { + t.Fatal("expected stopped") + } + if err := rt.Start(ctx); err != nil { + t.Fatal(err) + } + if rt.StartCount() != 2 { + t.Fatalf("restart did not relaunch worker: starts=%d", rt.StartCount()) + } + if rt.SubscriberCount() != 1 { + t.Fatalf("restart re-registered subscribers: %d", rt.SubscriberCount()) + } + _ = rt.Stop(ctx) + time.Sleep(10 * time.Millisecond) +} diff --git a/internal/app/chatflow/assembler.go b/internal/app/chatflow/assembler.go index e310bb7c..59254a43 100644 --- a/internal/app/chatflow/assembler.go +++ b/internal/app/chatflow/assembler.go @@ -70,16 +70,22 @@ func (a *Assembler) Assemble(ctx context.Context, p Principal, req AssembleReque if restored == nil { return userCtx, nil } - if restored.ProjectID > 0 && userCtx.Project.ID == "" { - userCtx.Project.ID = fmt.Sprintf("%d", restored.ProjectID) + upd := coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + BlackboardFill: restored.Facts, + SetRecentDialogue: true, + RecentDialogue: authorizedTurns(restored), } - if userCtx.BlackboardFacts == nil { - userCtx.BlackboardFacts = make(map[string]string) + if restored.ProjectID > 0 { + upd.ProjectIDIfEmpty = fmt.Sprintf("%d", restored.ProjectID) } - for k, v := range restored.Facts { - if _, exists := userCtx.BlackboardFacts[k]; !exists { - userCtx.BlackboardFacts[k] = v - } + _ = coordinator.ApplyContextUpdate(userCtx, upd) + return userCtx, nil +} + +func authorizedTurns(restored *ctxmgr.RestoredSessionContext) []coordinator.DialogueTurn { + if restored == nil { + return nil } turns := make([]coordinator.DialogueTurn, 0, len(restored.Messages)) records := make([]ctxmgr.ChatMessageRecord, 0, len(restored.Messages)) @@ -101,6 +107,5 @@ func (a *Assembler) Assemble(ctx context.Context, p Principal, req AssembleReque turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content}) } } - userCtx.RecentDialogue = turns - return userCtx, nil + return turns } diff --git a/internal/app/docnotify/subscriber.go b/internal/app/docnotify/subscriber.go new file mode 100644 index 00000000..2025e864 --- /dev/null +++ b/internal/app/docnotify/subscriber.go @@ -0,0 +1,94 @@ +package docnotify + +import ( + "context" + "fmt" + + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/docrevision" + "chat-agent-system/internal/notification" +) + +// DocumentLookup resolves a document for recipient and project fill-in. +type DocumentLookup interface { + GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error) +} + +// Subscriber is the unique consumer for document OnOutbox notification events. +// Bind once from the app composition root; do not assign OnOutbox in server. +type Subscriber struct { + Notify notificationEmitter + Docs DocumentLookup +} + +type notificationEmitter interface { + Available() bool + EmitDocumentEvent(ctx context.Context, eventType, notifType, title, body, dedupeKey string, recipientUserID int64, payload notification.DocumentEventPayload) (bool, error) +} + +// Bind assigns the subscriber exactly once. Repeat calls keep the first handler. +func Bind(rev *docrevision.Service, sub *Subscriber) bool { + if rev == nil || sub == nil || rev.OnOutbox != nil { + return false + } + rev.OnOutbox = sub.Handle + return true +} + +// Handle maps one document outbox event onto a permission-safe inbox notification. +func (s *Subscriber) Handle(ctx context.Context, eventType string, payload map[string]any) error { + if s == nil || s.Notify == nil || !s.Notify.Available() { + return nil + } + docID, _ := payload["doc_id"].(string) + projectID, _ := payload["project_id"].(int64) + risk, _ := payload["risk"].(string) + var recipient int64 + if s.Docs != nil && docID != "" { + if doc, err := s.Docs.GetDocument(ctx, docID); err == nil && doc != nil && doc.AuthorID > 0 { + recipient = doc.AuthorID + projectID = doc.ProjectID + } + } + if recipient == 0 { + return nil + } + p := notification.DocumentEventPayload{ + ProjectID: projectID, DocID: docID, RiskLevel: risk, + } + switch eventType { + case notification.EventDocAgentCandidateReady: + if v, ok := payload["candidate_id"].(int64); ok { + p.CandidateID = v + } + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentCandidateReady, + "智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。", + fmt.Sprintf("doc_candidate:%s:%d:%d", docID, p.CandidateID, recipient), recipient, p) + return err + case notification.EventDocAgentVersionCommitted: + if v, ok := payload["version_no"].(int); ok { + p.VersionNo = v + } + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentVersionCommitted, + "智能体版本已生效", "新的智能体文档版本已提交。", + fmt.Sprintf("doc_agent_version:%s:%d:%d", docID, p.VersionNo, recipient), recipient, p) + return err + case notification.EventDocDownstreamImpact: + impacted, _ := payload["impacted_doc_id"].(string) + p.ImpactedDocID = impacted + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact, + "下游文档受影响", "上游文档变更，请复核关联文档。", + fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p) + return err + case notification.EventDocChangeReportDegraded: + if v, ok := payload["report_id"].(int64); ok { + p.ReportID = v + } + _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocChangeReportDegraded, + "变更报告未完整评估", "变更报告降级，受控文档暂不可送审。", + fmt.Sprintf("doc_report_degraded:%s:%d:%d", docID, p.ReportID, recipient), recipient, p) + return err + default: + return nil + } +} diff --git a/internal/app/docnotify/subscriber_test.go b/internal/app/docnotify/subscriber_test.go new file mode 100644 index 00000000..16a6440f --- /dev/null +++ b/internal/app/docnotify/subscriber_test.go @@ -0,0 +1,73 @@ +package docnotify + +import ( + "context" + "testing" + + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/docrevision" + "chat-agent-system/internal/notification" +) + +type stubNotify struct { + available bool + calls int + lastType string +} + +func (s *stubNotify) Available() bool { return s != nil && s.available } + +func (s *stubNotify) EmitDocumentEvent(_ context.Context, eventType, _, _, _, _ string, _ int64, _ notification.DocumentEventPayload) (bool, error) { + s.calls++ + s.lastType = eventType + return true, nil +} + +type stubDocs struct { + doc *docgov.DocumentRecord +} + +func (s stubDocs) GetDocument(_ context.Context, _ string) (*docgov.DocumentRecord, error) { + return s.doc, nil +} + +func TestBindOnce(t *testing.T) { + rev := &docrevision.Service{} + sub := &Subscriber{Notify: &stubNotify{available: true}} + if !Bind(rev, sub) { + t.Fatal("first bind") + } + if Bind(rev, sub) { + t.Fatal("second bind must be a no-op") + } + if rev.OnOutbox == nil { + t.Fatal("hook missing") + } +} + +func TestHandleCandidateReady(t *testing.T) { + n := &stubNotify{available: true} + sub := &Subscriber{ + Notify: n, + Docs: stubDocs{doc: &docgov.DocumentRecord{AuthorID: 9, ProjectID: 4}}, + } + if err := sub.Handle(context.Background(), notification.EventDocAgentCandidateReady, map[string]any{ + "doc_id": "d1", "candidate_id": int64(3), + }); err != nil { + t.Fatal(err) + } + if n.calls != 1 || n.lastType != notification.EventDocAgentCandidateReady { + t.Fatalf("calls=%d type=%s", n.calls, n.lastType) + } +} + +func TestHandleSkipsUnknownAndNoRecipient(t *testing.T) { + n := &stubNotify{available: true} + sub := &Subscriber{Notify: n} + if err := sub.Handle(context.Background(), "document.unknown", map[string]any{"doc_id": "d1"}); err != nil { + t.Fatal(err) + } + if n.calls != 0 { + t.Fatalf("unknown event must not emit, calls=%d", n.calls) + } +} diff --git a/internal/app/stageflow/usecase.go b/internal/app/stageflow/usecase.go new file mode 100644 index 00000000..e0a34a82 --- /dev/null +++ b/internal/app/stageflow/usecase.go @@ -0,0 +1,40 @@ +package stageflow + +import ( + "context" + "errors" + + "chat-agent-system/internal/bizdev" +) + +var ErrClosed = errors.New("stageflow: executor not configured") + +// Executor is the unique application owner for task and stage runs. +// HTTP, CLI and coordinator must call this instead of bizdev directly. +type Executor interface { + ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) + ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) +} + +// UseCase adapts bizdev orchestration without moving domain tables. +type UseCase struct { + Biz Executor +} + +func New(biz Executor) *UseCase { + return &UseCase{Biz: biz} +} + +func (u *UseCase) ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) { + if u == nil || u.Biz == nil { + return nil, ErrClosed + } + return u.Biz.ExecuteTask(ctx, req) +} + +func (u *UseCase) ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) { + if u == nil || u.Biz == nil { + return nil, ErrClosed + } + return u.Biz.ExecuteStage(ctx, req) +} diff --git a/internal/app/stageflow/usecase_test.go b/internal/app/stageflow/usecase_test.go new file mode 100644 index 00000000..efe370d1 --- /dev/null +++ b/internal/app/stageflow/usecase_test.go @@ -0,0 +1,46 @@ +package stageflow + +import ( + "context" + "errors" + "testing" + + "chat-agent-system/internal/bizdev" +) + +type stubBiz struct { + tasks int + stages int +} + +func (s *stubBiz) ExecuteTask(_ context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) { + s.tasks++ + return &bizdev.TaskRunResult{Success: true, ArtifactTitle: req.TaskKey}, nil +} + +func (s *stubBiz) ExecuteStage(_ context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) { + s.stages++ + return &bizdev.StageRunResult{Success: true, StageTitle: string(req.Stage)}, nil +} + +func TestUseCaseDelegates(t *testing.T) { + biz := &stubBiz{} + httpUC := New(biz) + cliUC := New(biz) + if _, err := httpUC.ExecuteTask(context.Background(), bizdev.TaskRunRequest{TaskKey: "intake"}); err != nil { + t.Fatal(err) + } + if _, err := cliUC.ExecuteStage(context.Background(), bizdev.StageRunRequest{Stage: bizdev.StageV01_Triage}); err != nil { + t.Fatal(err) + } + if biz.tasks != 1 || biz.stages != 1 { + t.Fatalf("shared executor counts task=%d stage=%d", biz.tasks, biz.stages) + } +} + +func TestUseCaseClosed(t *testing.T) { + var u *UseCase + if _, err := u.ExecuteTask(context.Background(), bizdev.TaskRunRequest{}); !errors.Is(err, ErrClosed) { + t.Fatalf("got %v", err) + } +} diff --git a/internal/archtest/arch_test.go b/internal/archtest/arch_test.go index eb414381..8bce1e52 100644 --- a/internal/archtest/arch_test.go +++ b/internal/archtest/arch_test.go @@ -26,7 +26,25 @@ func TestArchitectureBoundaries(t *testing.T) { if ev.Failed() { t.Fatal(ev.Report()) } - if os.Getenv("ARCHTEST_STRICT") != "" && len(ev.Excepted) > 0 { - t.Fatalf("strict/block mode: %d baseline exceptions remain\n%s", len(ev.Excepted), ev.Report()) + if os.Getenv("ARCHTEST_STRICT") == "" { + return + } + if len(b.BlockKinds) == 0 { + if len(ev.Excepted) > 0 { + t.Fatalf("strict/block mode: %d baseline exceptions remain\n%s", len(ev.Excepted), ev.Report()) + } + return + } + for _, f := range ev.Excepted { + for _, kind := range b.BlockKinds { + if f.Kind == kind { + t.Fatalf("strict/block mode: cleared kind %s still has exception %s\n%s", kind, f.Key(), ev.Report()) + } + } + } + for _, kind := range b.BlockKinds { + if ev.Counts[kind] > 0 { + t.Fatalf("strict/block mode: cleared kind %s count=%d\n%s", kind, ev.Counts[kind], ev.Report()) + } } } diff --git a/internal/archtest/boundaries.go b/internal/archtest/boundaries.go index 8f575af3..3b8e2ecd 100644 --- a/internal/archtest/boundaries.go +++ b/internal/archtest/boundaries.go @@ -27,6 +27,7 @@ type Boundaries struct { UnwrapOwners []string `yaml:"unwrap_owners"` Exceptions []Exception `yaml:"exceptions"` BaselineCounts map[string]int `yaml:"baseline_counts"` + BlockKinds []string `yaml:"block_kinds"` } type Exception struct { diff --git a/internal/chat/session.go b/internal/chat/session.go index 85b99276..fae3289e 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -81,7 +81,11 @@ func NewSession(id string, userCtx *coordinator.UserProjectContext, rec intent.R } // 若上下文尚未挂起预期且当前已具有阶段，则初始化默认选项草稿 if userCtx != nil && userCtx.PendingExpectation == nil && userCtx.Project.CurrentStage != "" { - userCtx.PendingExpectation = coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + SetPendingExpectation: true, + PendingExpectation: coordinator.BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name), + }) } sess := &Session{ @@ -115,7 +119,11 @@ func (s *Session) SyncRecentDialogue() { } turns = append(turns, coordinator.DialogueTurn{Role: m.Role, Content: m.Content}) } - s.UserCtx.RecentDialogue = turns + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, + SetRecentDialogue: true, + RecentDialogue: turns, + }) // 若发生了截断且存在早期消息，同步计算或加载滚动摘要 if truncated && len(early) > 0 && s.CtxMgr != nil { @@ -416,11 +424,17 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str } if coordRes.ClearExpectation { - s.UserCtx.PendingExpectation = nil + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true, + }) } else if coordRes.NextExpectation != nil { - s.UserCtx.PendingExpectation = coordRes.NextExpectation + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true, PendingExpectation: coordRes.NextExpectation, + }) } else if recRes.Branch == intent.BranchExpectationHit { - s.UserCtx.PendingExpectation = nil + _ = coordinator.ApplyContextUpdate(s.UserCtx, coordinator.ContextUpdate{ + ExpectedVersion: s.UserCtx.Version, SetPendingExpectation: true, + }) } // 若为 BranchNewIntentPreempt 且未产生新预期，原挂起预期在焦点底静默保留，不阻拦用户 diff --git a/internal/coordinator/context.go b/internal/coordinator/context.go index 5b79e019..7472cee6 100644 --- a/internal/coordinator/context.go +++ b/internal/coordinator/context.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "time" "chat-agent-system/internal/bizdev" @@ -76,6 +77,8 @@ type DocumentSummaryReader func(ctx context.Context, target string) (*ReadableDo // UserProjectContext 用户项目级全量统一上下文 (对话系统与执行系统共同消费) type UserProjectContext struct { + mu sync.Mutex + Version int64 `json:"version"` SnapshotID string `json:"snapshot_id"` AssembledAt time.Time `json:"assembled_at"` User UserInfo `json:"user"` diff --git a/internal/coordinator/context_update.go b/internal/coordinator/context_update.go new file mode 100644 index 00000000..92ff07e8 --- /dev/null +++ b/internal/coordinator/context_update.go @@ -0,0 +1,144 @@ +package coordinator + +import ( + "errors" + + "chat-agent-system/internal/intent" +) + +var ( + ErrNilContext = errors.New("coordinator: nil user project context") + ErrStaleContext = errors.New("coordinator: stale context version") + ErrInvalidUpdate = errors.New("coordinator: empty context update") +) + +// ContextUpdate is the only typed write against UserProjectContext. +// Callers must supply ExpectedVersion from a prior read; a mismatch fails closed. +type ContextUpdate struct { + ExpectedVersion int64 + + BlackboardFill map[string]string + BlackboardSet map[string]string + + SetPendingExpectation bool + PendingExpectation *intent.PendingExpectation + ExpirePending bool + + SetInteractiveCard bool + InteractiveCard *intent.InteractiveCardPayload + TickInteractive bool + DismissCardStatus string // cancelled | resolved | expired; then nil + + SetPersonalizedGuidance bool + PersonalizedGuidance string + + SetRecentDialogue bool + RecentDialogue []DialogueTurn + + ProjectIDIfEmpty string +} + +// ApplyContextUpdate applies one versioned patch. Concurrent writers with the +// same ExpectedVersion: exactly one succeeds, the rest get ErrStaleContext. +func ApplyContextUpdate(ctx *UserProjectContext, u ContextUpdate) error { + if ctx == nil { + return ErrNilContext + } + ctx.mu.Lock() + defer ctx.mu.Unlock() + if u.ExpectedVersion != ctx.Version { + return ErrStaleContext + } + applied := false + if len(u.BlackboardFill) > 0 { + if ctx.BlackboardFacts == nil { + ctx.BlackboardFacts = make(map[string]string, len(u.BlackboardFill)) + } + for k, v := range u.BlackboardFill { + if _, exists := ctx.BlackboardFacts[k]; !exists { + ctx.BlackboardFacts[k] = v + } + } + applied = true + } + if len(u.BlackboardSet) > 0 { + if ctx.BlackboardFacts == nil { + ctx.BlackboardFacts = make(map[string]string, len(u.BlackboardSet)) + } + for k, v := range u.BlackboardSet { + ctx.BlackboardFacts[k] = v + } + applied = true + } + if u.ExpirePending && ctx.PendingExpectation != nil && !ctx.PendingExpectation.ExpiresAt.IsZero() { + // caller decides freshness; apply just clears + ctx.PendingExpectation = nil + applied = true + } + if u.SetPendingExpectation { + ctx.PendingExpectation = u.PendingExpectation + applied = true + } + if u.TickInteractive && ctx.InteractiveCard != nil { + ctx.InteractiveCard.TTLTurns-- + if ctx.InteractiveCard.TTLTurns <= 0 || ctx.InteractiveCard.CurrentStatus != "active" { + ctx.InteractiveCard.CurrentStatus = "expired" + ctx.InteractiveCard = nil + } + applied = true + } + if u.DismissCardStatus != "" && ctx.InteractiveCard != nil { + ctx.InteractiveCard.CurrentStatus = u.DismissCardStatus + ctx.InteractiveCard = nil + applied = true + } + if u.SetInteractiveCard { + ctx.InteractiveCard = u.InteractiveCard + applied = true + } + if u.SetPersonalizedGuidance { + ctx.PersonalizedGuidance = u.PersonalizedGuidance + applied = true + } + if u.SetRecentDialogue { + ctx.RecentDialogue = u.RecentDialogue + applied = true + } + if u.ProjectIDIfEmpty != "" && ctx.Project.ID == "" { + ctx.Project.ID = u.ProjectIDIfEmpty + applied = true + } + if !applied { + return ErrInvalidUpdate + } + ctx.Version++ + return nil +} + +func (c *UserProjectContext) applyPending(exp *intent.PendingExpectation) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, SetPendingExpectation: true, PendingExpectation: exp}) +} + +func (c *UserProjectContext) applyCard(card *intent.InteractiveCardPayload) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, SetInteractiveCard: true, InteractiveCard: card}) +} + +func (c *UserProjectContext) dismissCard(status string) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, DismissCardStatus: status}) +} + +func (c *UserProjectContext) applyBlackboard(key, value string) { + if c == nil { + return + } + _ = ApplyContextUpdate(c, ContextUpdate{ExpectedVersion: c.Version, BlackboardSet: map[string]string{key: value}}) +} diff --git a/internal/coordinator/context_update_test.go b/internal/coordinator/context_update_test.go new file mode 100644 index 00000000..cb7aa537 --- /dev/null +++ b/internal/coordinator/context_update_test.go @@ -0,0 +1,55 @@ +package coordinator + +import ( + "errors" + "sync" + "sync/atomic" + "testing" +) + +func TestContextUpdateConcurrentNoLostUpdate(t *testing.T) { + ctx := &UserProjectContext{ + Version: 1, + BlackboardFacts: map[string]string{"market": "国内"}, + } + var ok, stale int32 + var wg sync.WaitGroup + for i := 0; i < 2; i++ { + wg.Add(1) + go func(label string) { + defer wg.Done() + err := ApplyContextUpdate(ctx, ContextUpdate{ + ExpectedVersion: 1, + BlackboardSet: map[string]string{"market": label}, + }) + switch { + case err == nil: + atomic.AddInt32(&ok, 1) + case errors.Is(err, ErrStaleContext): + atomic.AddInt32(&stale, 1) + default: + t.Errorf("unexpected %v", err) + } + }("v" + string(rune('a'+i))) + } + wg.Wait() + if ok != 1 || stale != 1 { + t.Fatalf("ok=%d stale=%d version=%d facts=%v", ok, stale, ctx.Version, ctx.BlackboardFacts) + } + if ctx.Version != 2 { + t.Fatalf("version %d", ctx.Version) + } + if ctx.BlackboardFacts["market"] == "国内" { + t.Fatal("winner did not apply") + } +} + +func TestContextUpdateFillDoesNotOverwrite(t *testing.T) { + ctx := &UserProjectContext{Version: 0, BlackboardFacts: map[string]string{"a": "1"}} + if err := ApplyContextUpdate(ctx, ContextUpdate{ExpectedVersion: 0, BlackboardFill: map[string]string{"a": "x", "b": "2"}}); err != nil { + t.Fatal(err) + } + if ctx.BlackboardFacts["a"] != "1" || ctx.BlackboardFacts["b"] != "2" { + t.Fatalf("%v", ctx.BlackboardFacts) + } +} diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index 97301339..82a676be 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -73,12 +73,36 @@ type CoordinationResult struct { // CoordinatorEngine L2 协调引擎 type CoordinatorEngine struct { bizdevSvc bizdev.BizdevService + stageTasks StageTaskRunner instSvc *projectinstance.Service assembler *ContextAssembler dialogueBackend agent.Backend dataDir string } +// StageTaskRunner is the application owner for ExecuteTask / ExecuteStage. +// Wired from app/stageflow; tests may leave it nil and fall back to bizdevSvc. +type StageTaskRunner interface { + ExecuteTask(ctx context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) + ExecuteStage(ctx context.Context, req bizdev.StageRunRequest) (*bizdev.StageRunResult, error) +} + +func (e *CoordinatorEngine) SetStageTaskRunner(r StageTaskRunner) { + if e != nil { + e.stageTasks = r + } +} + +func (e *CoordinatorEngine) tasks() StageTaskRunner { + if e == nil { + return nil + } + if e.stageTasks != nil { + return e.stageTasks + } + return e.bizdevSvc +} + // SetInstanceService 注入项目工作流实例服务（对话推进与实例门禁合流）。 func (e *CoordinatorEngine) SetInstanceService(svc *projectinstance.Service) { if e != nil { @@ -118,14 +142,10 @@ func (e *CoordinatorEngine) ProcessIntent( // 交互卡片 TTL 轮次消退与生命周期清理 if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.TTLTurns-- - if userCtx.InteractiveCard.TTLTurns <= 0 || userCtx.InteractiveCard.CurrentStatus != "active" { - userCtx.InteractiveCard.CurrentStatus = "expired" - userCtx.InteractiveCard = nil - } + _ = ApplyContextUpdate(userCtx, ContextUpdate{ExpectedVersion: userCtx.Version, TickInteractive: true}) } if userCtx.PendingExpectation != nil && !userCtx.PendingExpectation.ExpiresAt.IsZero() && time.Now().After(userCtx.PendingExpectation.ExpiresAt) { - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) } top := recRes.TopIntent @@ -355,10 +375,9 @@ func (e *CoordinatorEngine) handleDeclineExpectation(userCtx *UserProjectContext if projName == "" { projName = "当前项目" } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "cancelled" - userCtx.InteractiveCard = nil + userCtx.dismissCard("cancelled") } reply := fmt.Sprintf("收到，听您的！那我们先不急着推进。\n针对「%s」，您手头还有哪些细节、技术顾虑或商务考量想先聊聊？我随时待命为您解答。", projName) @@ -418,7 +437,7 @@ func (e *CoordinatorEngine) handleAmbiguousConsult(ctx context.Context, cand *in nextExp := BuildNextExpectation(currentStage, projName) // 路由层：复杂自由任务画像命中时，先给即时初答，再附升级建议（复用挂起期望机制，含每会话冷却） reply, nextExp = maybeAttachUpgradeHint(rawInput, userCtx, reply, nextExp) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) return &CoordinationResult{ Status: StatusAccepted, @@ -436,7 +455,7 @@ func (e *CoordinatorEngine) handleQueryProgress(ctx context.Context, cand *inten brief := userCtx.LatestAdvisorBrief topic := cand.FilledSlots[intent.SlotQueryTopic] nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) // 1. 若为测试方案/难点/验证咨询，直接调取测试经理与参谋专业知识库解答 if topic == "testing_analysis" { @@ -624,7 +643,7 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u summaryKey := "" if factKey != "" && factVal != "" { - userCtx.BlackboardFacts[factKey] = factVal + userCtx.applyBlackboard(factKey, factVal) summaryKey = fmt.Sprintf("「%s」已更新为「%s」", factKey, factVal) if factKey == "target_market" { userCtx.Project.TargetMarket = factVal @@ -657,7 +676,7 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u } nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) var sb strings.Builder sb.WriteString(fmt.Sprintf("已收到并为您记录最新信息：%s。\n", summaryKey)) @@ -757,7 +776,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent if len(cand.PatchSlots) > 0 { var patches []string for k, v := range cand.PatchSlots { - userCtx.BlackboardFacts[k] = v + userCtx.applyBlackboard(k, v) patches = append(patches, fmt.Sprintf("%s=%s", k, v)) if k == "target_market" { userCtx.Project.TargetMarket = v @@ -795,11 +814,11 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent }, nil } - if e.bizdevSvc == nil { + if e.tasks() == nil { userCtx.Project.CurrentStage = targetStage userCtx.AvailableActions = ComputeAvailableActions(targetStage, userCtx.User.ProjectRole) nextExp := BuildNextExpectation(targetStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) reply := fmt.Sprintf("已为您模拟推进到阶段【%s】。", targetStage) if patchSummary != "" { @@ -816,7 +835,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent } // 3. 调用底层 L3 调度器执行阶段跃迁 - stageRes, err := e.bizdevSvc.ExecuteStage(ctx, bizdev.StageRunRequest{ + stageRes, err := e.tasks().ExecuteStage(ctx, bizdev.StageRunRequest{ Stage: targetStage, ProjectName: userCtx.Project.Name, CustomerName: userCtx.Project.CustomerName, @@ -850,7 +869,7 @@ func (e *CoordinatorEngine) handleAdvanceStage(ctx context.Context, cand *intent } nextExp := BuildNextExpectation(targetStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) var sb strings.Builder if patchSummary != "" { @@ -889,9 +908,9 @@ func (e *CoordinatorEngine) handleTriggerTask(ctx context.Context, cand *intent. } nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) - if e.bizdevSvc == nil { + if e.tasks() == nil { return &CoordinationResult{ Status: StatusAccepted, Intent: cand.Intent, @@ -910,7 +929,7 @@ func (e *CoordinatorEngine) handleTriggerTask(ctx context.Context, cand *intent. fmt.Sscanf(userCtx.User.ID, "%d", &userID) } - taskRes, err := e.bizdevSvc.ExecuteTask(ctx, bizdev.TaskRunRequest{ + taskRes, err := e.tasks().ExecuteTask(ctx, bizdev.TaskRunRequest{ TaskKey: taskKey, ProjectName: userCtx.Project.Name, CustomerName: userCtx.Project.CustomerName, @@ -1026,9 +1045,9 @@ func (e *CoordinatorEngine) handleTriggerAdhocTask(ctx context.Context, cand *in taskKey := sanitizeAdhocTaskKey(title) nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) - if e.bizdevSvc == nil { + if e.tasks() == nil { return &CoordinationResult{ Status: StatusAccepted, Intent: cand.Intent, @@ -1058,7 +1077,7 @@ func (e *CoordinatorEngine) handleTriggerAdhocTask(ctx context.Context, cand *in Content: []byte(fmt.Sprintf("# 任务简报（用户原始诉求，权威等级：primary_evidence）\n\n%s\n", brief)), } - taskRes, err := e.bizdevSvc.ExecuteTask(ctx, bizdev.TaskRunRequest{ + taskRes, err := e.tasks().ExecuteTask(ctx, bizdev.TaskRunRequest{ TaskKey: taskKey, Title: title, ProjectName: userCtx.Project.Name, @@ -1141,7 +1160,7 @@ func (e *CoordinatorEngine) handleCasualChat(ctx context.Context, userCtx *UserP } nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) nameSuffix := "" if userCtx.User.Name != "" { @@ -1157,7 +1176,7 @@ func (e *CoordinatorEngine) handleCasualChat(ctx context.Context, userCtx *UserP // 路由层：复杂自由任务画像命中时，先给即时初答，再附升级建议（复用挂起期望机制，含每会话冷却） reply, nextExp = maybeAttachUpgradeHint(rawInput, userCtx, reply, nextExp) - userCtx.PendingExpectation = nextExp + userCtx.applyPending(nextExp) return &CoordinationResult{ Status: StatusAccepted, @@ -1254,8 +1273,8 @@ func (e *CoordinatorEngine) handleDocGovReject(ctx context.Context, cand *intent }, nil } - userCtx.InteractiveCard = resolved.Card - userCtx.PendingExpectation = resolved.Expectation + userCtx.applyCard(resolved.Card) + userCtx.applyPending(resolved.Expectation) return &CoordinationResult{ Status: StatusAccepted, @@ -1299,7 +1318,7 @@ func (e *CoordinatorEngine) handleQueryRejectReason(ctx context.Context, cand *i resolved := ResolveRejectCard(role, "", docName, docName, "", reason, RejectUtteranceQuery) if resolved.Card != nil { - userCtx.InteractiveCard = resolved.Card + userCtx.applyCard(resolved.Card) } return &CoordinationResult{ @@ -1442,8 +1461,8 @@ func (e *CoordinatorEngine) handleDocGovApprove(ctx context.Context, cand *inten }, } - userCtx.InteractiveCard = cardPayload - userCtx.PendingExpectation = nextExp + userCtx.applyCard(cardPayload) + userCtx.applyPending(nextExp) reply := fmt.Sprintf("已为您起草针对《%s》的终审批准确认单。终审批准将使文档正式归档并永久锁定，请审阅下方卡片并在确认无误后点击确认或回复‘确认’。", docName) @@ -1559,10 +1578,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand } if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "resolved" - userCtx.InteractiveCard = nil + userCtx.dismissCard("resolved") } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) rollback := docgov.StateDrafting if userCtx.InteractiveCard != nil && strings.Contains(userCtx.InteractiveCard.ImpactRisk, "reviewing") { @@ -1587,10 +1605,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand if targetAction == "confirm_final_approve" { if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "resolved" - userCtx.InteractiveCard = nil + userCtx.dismissCard("resolved") } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) return &CoordinationResult{ Status: StatusAccepted, @@ -1608,10 +1625,9 @@ func (e *CoordinatorEngine) handleDocGovProposalDecide(ctx context.Context, cand if targetAction == "cancel" { if userCtx.InteractiveCard != nil { - userCtx.InteractiveCard.CurrentStatus = "cancelled" - userCtx.InteractiveCard = nil + userCtx.dismissCard("cancelled") } - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) return &CoordinationResult{ Status: StatusAccepted, diff --git a/internal/coordinator/engine_run.go b/internal/coordinator/engine_run.go index ddfb2cca..b52883af 100644 --- a/internal/coordinator/engine_run.go +++ b/internal/coordinator/engine_run.go @@ -50,7 +50,7 @@ func (e *CoordinatorEngine) handleRunResume(_ context.Context, cand *intent.Cand } if isRunActionConfirmed(userCtx, "run.resume") { - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) return &CoordinationResult{ Status: StatusAccepted, Intent: intent.IntentRunResume, @@ -96,7 +96,7 @@ func (e *CoordinatorEngine) handleRunResume(_ context.Context, cand *intent.Cand func (e *CoordinatorEngine) handleRunRestart(_ context.Context, cand *intent.CandidateIntent, userCtx *UserProjectContext) (*CoordinationResult, error) { if isRunActionConfirmed(userCtx, "run.restart") { - userCtx.PendingExpectation = nil + userCtx.applyPending(nil) taskKey := cand.FilledSlots["task_key"] if taskKey == "" && userCtx.TaskState != nil { taskKey = userCtx.TaskState.ActiveTask diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 4de4ff31..ef183972 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -253,7 +253,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { // 1. 如果指定了 stage，则执行成熟度阶段编排 if req.Stage != "" { - stageRes, err := s.svc.ExecuteStage(r.Context(), bizdev.StageRunRequest{ + stageRes, err := s.taskUseCase().ExecuteStage(r.Context(), bizdev.StageRunRequest{ Stage: bizdev.MaturityStage(req.Stage), ProjectName: req.ProjectName, CustomerName: req.CustomerName, @@ -301,7 +301,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { if req.TaskKey == "" { req.TaskKey = "intake" } - taskRes, err := s.svc.ExecuteTask(r.Context(), bizdev.TaskRunRequest{ + taskRes, err := s.taskUseCase().ExecuteTask(r.Context(), bizdev.TaskRunRequest{ TaskKey: req.TaskKey, ProjectName: req.ProjectName, CustomerName: req.CustomerName, diff --git a/internal/server/server.go b/internal/server/server.go index 666199d0..ca1ee34c 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -10,10 +10,9 @@ import ( "sync" "time" - "chat-agent-system/internal/agent" + "chat-agent-system/internal/app" "chat-agent-system/internal/app/chatflow" - "chat-agent-system/internal/app/uow" - "chat-agent-system/internal/artifactvault" + "chat-agent-system/internal/app/stageflow" "chat-agent-system/internal/auth" "chat-agent-system/internal/authz" "chat-agent-system/internal/bizdev" @@ -23,7 +22,6 @@ import ( "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/intent" "chat-agent-system/internal/notification" - "chat-agent-system/internal/observability" "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/taskrun" @@ -33,6 +31,7 @@ import ( // Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务 type Server struct { + rt *app.Runtime cfg *config.Config svc *bizdev.Service pwSvc *projectworkflow.Service @@ -52,156 +51,55 @@ type Server struct { cmds *taskrun.Commands chatLog chatflow.ChatLog assembler chatflow.ContextAssembler + stageflow *stageflow.UseCase httpServer *http.Server } -// NewServer 构造函数 -func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { - if cfg == nil { - cfg, _ = config.Load("") +// NewFromRuntime attaches the HTTP adapter to an already-wired composition root. +func NewFromRuntime(rt *app.Runtime) *Server { + if rt == nil { + panic("server: nil runtime") } - if cfg != nil { - observability.InitLogger(cfg.Env) - } - if svc == nil { - var err error - svc, err = bizdev.NewService(cfg) - if err != nil { - panic("server: bizdev service init failed: " + err.Error()) - } - } - - rec := intent.NewRecognizer() - dialogueBackend := agent.NewBackend(cfg.AgentBackend, cfg) - coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir) - pool := svc.Pool() - stores, storeErr := auth.NewStores(context.Background(), pool, cfg) - if storeErr != nil { - panic("server: auth store init failed: " + storeErr.Error()) - } - authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg) - pwSvc := svc.ProjectWorkflow() - if pwSvc == nil { - pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil) - } - - authzSvc := authz.NewService(pool) - wfDefSvc := workflowdef.NewService(pool, svc.CAS()) - instStore := projectinstance.NewStore(pool) - var instSvc *projectinstance.Service - if wfDefSvc.Available() { - docChecker := projectinstance.NewDocGovChecker(svc.DocGov()) - instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker) - if pool != nil { - instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool)) - } - } - taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager()) - 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, - svc: svc, - pwSvc: pwSvc, - rec: rec, - engine: coordEngine, - authSvc: authSvc, + rt: rt, + cfg: rt.Cfg, + svc: rt.Biz, + pwSvc: rt.PW, + rec: rt.Rec, + engine: rt.Engine, + authSvc: rt.Auth, sessions: make(map[string]*chat.Session), - taskRuns: taskMgr, - authorizer: authz.NewAuthorizer(pool), - authzSvc: authzSvc, - wfDefSvc: wfDefSvc, - instSvc: instSvc, - notifySvc: notification.NewService(pool), - resolver: workspace.NewResolver(cfg.DataDir, pool), - wsCap: wsCap, - cmds: cmds, + taskRuns: rt.TaskRuns, + authorizer: rt.Authorizer, + authzSvc: rt.Authz, + wfDefSvc: rt.WFDef, + instSvc: rt.Inst, + notifySvc: rt.Notify, + resolver: rt.Resolver, + wsCap: rt.WSCap, + cmds: rt.Commands, + stageflow: rt.Stageflow, } - s.chatLog = chatflow.NewAdapter(svc.ContextManager(), s.resolveStageAccess, s.failClosedAuthz()) + s.chatLog = chatflow.NewAdapter(s.svc.ContextManager(), s.resolveStageAccess, s.failClosedAuthz()) s.assembler = chatflow.NewAssembler(coordinator.NewContextAssembler(), s.chatLog) - if instSvc != nil && instSvc.Available() { - svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc)) - coordEngine.SetInstanceService(instSvc) + rt.ChatLog = s.chatLog + rt.Assembler = s.assembler + if s.instSvc != nil && s.instSvc.Available() && s.svc != nil { + s.svc.SetStageAccessGate(newBizdevStageGate(s.instSvc, s.pwSvc)) } - s.registerWorkflowApprovalHook() - s.startNotificationReminderLoop() s.initTaskRunExecutors() - if svc != nil && svc.DocRevision() != nil && s.notifySvc != nil { - rev := svc.DocRevision() - rev.OnOutbox = func(ctx context.Context, eventType string, payload map[string]any) error { - if !s.notifySvc.Available() { - return nil - } - docID, _ := payload["doc_id"].(string) - projectID, _ := payload["project_id"].(int64) - risk, _ := payload["risk"].(string) - var recipient int64 - if s.svc.DocGov() != nil { - if doc, err := s.svc.DocGov().GetDocument(ctx, docID); err == nil && doc != nil && doc.AuthorID > 0 { - recipient = doc.AuthorID - projectID = doc.ProjectID - } - } - if recipient == 0 { - return nil - } - p := notification.DocumentEventPayload{ - ProjectID: projectID, DocID: docID, RiskLevel: risk, - } - switch eventType { - case notification.EventDocAgentCandidateReady: - if v, ok := payload["candidate_id"].(int64); ok { - p.CandidateID = v - } - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentCandidateReady, - "智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。", - fmt.Sprintf("doc_candidate:%s:%d:%d", docID, p.CandidateID, recipient), recipient, p) - return err - case notification.EventDocAgentVersionCommitted: - if v, ok := payload["version_no"].(int); ok { - p.VersionNo = v - } - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentVersionCommitted, - "智能体版本已生效", "新的智能体文档版本已提交。", - fmt.Sprintf("doc_agent_version:%s:%d:%d", docID, p.VersionNo, recipient), recipient, p) - return err - case notification.EventDocDownstreamImpact: - impacted, _ := payload["impacted_doc_id"].(string) - p.ImpactedDocID = impacted - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact, - "下游文档受影响", "上游文档变更，请复核关联文档。", - fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p) - return err - case notification.EventDocChangeReportDegraded: - if v, ok := payload["report_id"].(int64); ok { - p.ReportID = v - } - _, err := s.notifySvc.EmitDocumentEvent(ctx, eventType, notification.TypeDocChangeReportDegraded, - "变更报告未完整评估", "变更报告降级，受控文档暂不可送审。", - fmt.Sprintf("doc_report_degraded:%s:%d:%d", docID, p.ReportID, recipient), recipient, p) - return err - default: - return nil - } - } - } return s } +// NewServer 构造函数（测试兼容：内部走 app.Assemble） +func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { + rt, err := app.Assemble(cfg, svc) + if err != nil { + panic("server: runtime init failed: " + err.Error()) + } + return NewFromRuntime(rt) +} + // AuthService 获取认证服务 func (s *Server) AuthService() *auth.Service { return s.authSvc @@ -426,6 +324,9 @@ func (s *Server) serveSPA(w http.ResponseWriter, r *http.Request) { // Start 启动 HTTP 守护进程 func (s *Server) Start() error { + if s.rt != nil { + _ = s.rt.Start(context.Background()) + } addr := fmt.Sprintf("%s:%d", s.cfg.HTTPHost, s.cfg.HTTPPort) s.httpServer = &http.Server{ Addr: addr, @@ -441,6 +342,9 @@ func (s *Server) Start() error { // Shutdown 优雅停机 func (s *Server) Shutdown(ctx context.Context) error { + if s.rt != nil { + _ = s.rt.Stop(ctx) + } if s.httpServer != nil { return s.httpServer.Shutdown(ctx) } @@ -558,11 +462,18 @@ func (s *Server) getOrCreateSession( BlackboardFacts: make(map[string]string), AvailableActions: coordinator.ComputeAvailableActions(bizdev.StageV01_Triage, u.ProjectRole), } - if targetMarket != "" { - userCtx.BlackboardFacts["target_market"] = targetMarket - } - if deviceType != "" { - userCtx.BlackboardFacts["device_type"] = deviceType + if targetMarket != "" || deviceType != "" { + fill := map[string]string{} + if targetMarket != "" { + fill["target_market"] = targetMarket + } + if deviceType != "" { + fill["device_type"] = deviceType + } + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + BlackboardFill: fill, + }) } } @@ -586,10 +497,11 @@ func (s *Server) getOrCreateSession( if restored.ProjectID > 0 { sess.ProjectID = restored.ProjectID } - for k, v := range restored.Facts { - if _, exists := userCtx.BlackboardFacts[k]; !exists { - userCtx.BlackboardFacts[k] = v - } + if len(restored.Facts) > 0 { + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + BlackboardFill: restored.Facts, + }) } if s.chatLog == nil { for _, rm := range restored.Messages { @@ -607,7 +519,11 @@ func (s *Server) getOrCreateSession( if sess.UserID > 0 && sess.CtxMgr != nil { if guidance, err := sess.CtxMgr.CompilePromptGuidance(context.Background(), sess.UserID); err == nil && guidance != "" { - userCtx.PersonalizedGuidance = guidance + _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ + ExpectedVersion: userCtx.Version, + SetPersonalizedGuidance: true, + PersonalizedGuidance: guidance, + }) } } } @@ -641,6 +557,16 @@ func (s *Server) hydrateVisibleHistory(ctx context.Context, sess *chat.Session, sess.SyncRecentDialogue() } +func (s *Server) taskUseCase() *stageflow.UseCase { + if s == nil { + return stageflow.New(nil) + } + if s.stageflow != nil { + return s.stageflow + } + return stageflow.New(s.svc) +} + func (s *Server) corsMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { s.writeCORS(w, r) diff --git a/internal/server/sse.go b/internal/server/sse.go index 921fcad3..171dbe2b 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -13,6 +13,7 @@ import ( "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/squad" "chat-agent-system/internal/stream" + "chat-agent-system/internal/stream/transport" "chat-agent-system/internal/taskrun" ) @@ -55,7 +56,7 @@ func extendSSEWriteDeadline(w http.ResponseWriter) { func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error { extendSSEWriteDeadline(w) - if _, err := fmt.Fprintf(w, ": heartbeat %d\n\n", time.Now().Unix()); err != nil { + if err := (transport.Encoder{W: w}).Heartbeat(time.Now()); err != nil { return err } flusher.Flush() @@ -63,11 +64,7 @@ func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error { } func writeSSE(w http.ResponseWriter, flusher http.Flusher, ev stream.StreamEvent) { - data, err := json.Marshal(ev) - if err != nil { - return - } - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + _ = (transport.Encoder{W: w}).Write(ev) flusher.Flush() } @@ -236,14 +233,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { listener := func(ev stream.StreamEvent) { writeMu.Lock() defer writeMu.Unlock() - - data, err := json.Marshal(ev) - if err != nil { - return - } - // 按照标准 SSE 协议规范发射: event: \ndata: \n\n - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, string(data)) - flusher.Flush() + writeSSE(w, flusher, ev) } clientTurnID := req.ClientTurnID @@ -302,15 +292,12 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) { if err != nil { writeMu.Lock() info := stream.ClassifyError(err, "chat_stream", requestID, sess.ID, "") - errEv := stream.StreamEvent{ + writeSSE(w, flusher, stream.StreamEvent{ Type: stream.EventError, Timestamp: time.Now(), Error: info.UserMessage, ErrorInfo: &info, - } - data, _ := json.Marshal(errEv) - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", errEv.Type, string(data)) - flusher.Flush() + }) writeMu.Unlock() } } @@ -408,12 +395,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { defer s.taskRuns.Unsubscribe(created.RunID, ch) writeStored := func(stored taskrun.StoredEvent) { - data, marshalErr := json.Marshal(stored.Event) - if marshalErr != nil { - return - } - fmt.Fprintf(w, "id: %d\n", stored.Seq) - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", stored.Event.Type, string(data)) + _ = (transport.Encoder{W: w}).WriteWithID(stored.Seq, stored.Event) flusher.Flush() } diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index a25babf6..675361a2 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -708,7 +708,7 @@ func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clien func (s *Server) initTaskRunExecutors() { s.taskRuns.SetExecutor( func(ctx context.Context, _ string, req bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { - return s.svc.ExecuteTask(ctx, req) + return s.taskUseCase().ExecuteTask(ctx, req) }, func(ctx context.Context, _ string, workDir string, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { res, err := s.svc.ResumeOrModifyWorkspace(ctx, squad.ModifyWorkflowRequest{WorkDir: workDir}) @@ -722,7 +722,7 @@ func (s *Server) initTaskRunExecutors() { return out, err }, func(ctx context.Context, _ string, req bizdev.StageRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { - stageRes, err := s.svc.ExecuteStage(ctx, req) + stageRes, err := s.taskUseCase().ExecuteStage(ctx, req) if stageRes == nil { return nil, err } diff --git a/internal/stream/transport/sse.go b/internal/stream/transport/sse.go new file mode 100644 index 00000000..0ad76d40 --- /dev/null +++ b/internal/stream/transport/sse.go @@ -0,0 +1,47 @@ +package transport + +import ( + "encoding/json" + "fmt" + "io" + "time" + + "chat-agent-system/internal/stream" +) + +// Encoder writes neutral SSE frames. It does not own retries, leases, or task state. +type Encoder struct { + W io.Writer +} + +func (e Encoder) Write(ev stream.StreamEvent) error { + return e.WriteWithID(0, ev) +} + +func (e Encoder) WriteWithID(id int64, ev stream.StreamEvent) error { + if e.W == nil { + return nil + } + if ev.Timestamp.IsZero() { + ev.Timestamp = time.Now() + } + data, err := json.Marshal(ev) + if err != nil { + return err + } + if id > 0 { + if _, err = fmt.Fprintf(e.W, "id: %d\n", id); err != nil { + return err + } + } + _, err = fmt.Fprintf(e.W, "event: %s\ndata: %s\n\n", ev.Type, string(data)) + return err +} + +func (e Encoder) Heartbeat(now time.Time) error { + if e.W == nil { + return nil + } + _, err := fmt.Fprintf(e.W, ": heartbeat %d\n\n", now.Unix()) + return err +} diff --git a/internal/stream/transport/sse_test.go b/internal/stream/transport/sse_test.go new file mode 100644 index 00000000..a615edff --- /dev/null +++ b/internal/stream/transport/sse_test.go @@ -0,0 +1,35 @@ +package transport + +import ( + "bytes" + "strings" + "testing" + "time" + + "chat-agent-system/internal/stream" +) + +func TestWriteNeutralFrame(t *testing.T) { + var buf bytes.Buffer + enc := Encoder{W: &buf} + if err := enc.Write(stream.StreamEvent{Type: stream.EventDone, Timestamp: time.Unix(1, 0)}); err != nil { + t.Fatal(err) + } + got := buf.String() + if !strings.Contains(got, "event: done") || !strings.Contains(got, "data: {") { + t.Fatalf("%q", got) + } + if strings.Contains(got, "retry:") { + t.Fatal("transport must not own retry") + } +} + +func TestHeartbeat(t *testing.T) { + var buf bytes.Buffer + if err := (Encoder{W: &buf}).Heartbeat(time.Unix(10, 0)); err != nil { + t.Fatal(err) + } + if !strings.Contains(buf.String(), ": heartbeat 10") { + t.Fatalf("%q", buf.String()) + } +} diff --git a/scripts/arch-check.sh b/scripts/arch-check.sh index fc82c2a2..4ceb73b0 100755 --- a/scripts/arch-check.sh +++ b/scripts/arch-check.sh @@ -31,7 +31,7 @@ fi if [[ "${MODE}" == "block" ]]; then if ! ARCHTEST_STRICT=1 go test ./internal/archtest -count=1 -run TestArchitectureBoundaries; then - echo "arch-check block: baseline exceptions remain; only cleared rules may use --mode block" >&2 + echo "arch-check block: a cleared kind in block_kinds still has findings" >&2 exit 1 fi fi