diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 1d41fa2e..2e53c00f 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -5,7 +5,7 @@ ## 焦点 -1. **TRACK-E/S**: Phase 2 隔离库与结构解耦已合入;共享 live 见 [长期任务](./long-term/2026-09-15-event-shared-live-cutover.md) +1. **TRACK-E LIVE**: 共享 consumer 切换([short-term](./short-term/2026-09-15-event-shared-live-cutover.md)) 2. **TRACK-P**: 立项完成([H0](./research/2026-09-15-http-error-envelope-h0-inventory.md) / [P0](./research/2026-09-15-pi-typed-result-p0-matrix.md));实施见 [长期任务](./long-term/2026-09-15-track-p-protocol-implementation.md) 3. **CHAT-INTENT**: [对话意图运行时状态](../test-reports/2026-09-13-conversation-intent-runtime-state-test-report.md)(双标签 busy 已签:项目 **6275** / `DOC_BUSY` 单 run;不删 `sess-p143-u9` 探针) @@ -70,6 +70,7 @@ | Cursor | bizdev-locator-removal | fix/bizdev-locator-removal | - | bizdev | done | | Cursor | chatflow-orchestration-boundary | fix/chatflow-orchestration-boundary | - | chat | done | | Cursor | phase2-closeout-inventories | feature/phase2-closeout-inventories | - | architecture-docs | done | +| Cursor | event-shared-live-cutover | feature/event-shared-live-cutover | - | events | in_progress | > 新任务开始前登记。`branch` 必须为 `feature/` 或 `fix/`,**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。 @@ -81,7 +82,7 @@ | holder | phase | since | scope | queue | |--------|-------|-------|-------|-------| -| - | idle | - | - | - | +| Cursor | e2e | 2026-09-15 | shared event consumer live cutover;FAV off | - | **phase**: `idle` | `merging` | `smoke` | `e2e` diff --git a/.doc/ops/short-term/2026-09-15-event-shared-live-cutover.md b/.doc/ops/short-term/2026-09-15-event-shared-live-cutover.md new file mode 100644 index 00000000..4b5ee859 --- /dev/null +++ b/.doc/ops/short-term/2026-09-15-event-shared-live-cutover.md @@ -0,0 +1,50 @@ +--- +title: "共享环境事件消费者 live 切换" +status: in_progress +created: 2026-09-15 +updated: 2026-09-15 +owner: Cursor +branch: feature/event-shared-live-cutover +worktree: "-" +mutex: events +--- + +# 共享环境事件消费者 live 切换 + +## 目标 + +在共享 `agent-dev` 上按 notify → workflow_advance → projection 完成 shadow → drain → live → 回滚演练,再删除生产 hook 注册。 + +## 范围 + +- `scripts/event-consumer-mode.sh` + `cmd/eventmode` +- 共享 `agent_system_dev` inventory / quarantine +- 逐消费者 live 与授权页纵切(项目 6275;143 只读) + +## 非目标 + +- FAV live +- test/main 环境 +- 编辑 Phase 2 计划原文 +- Track P(独立波次) + +## 步骤 + +1. [x] L0 受控 SetMode 与 inventory +2. [ ] L1 notify +3. [ ] L2 workflow_advance +4. [ ] L3 projection + 授权页 +5. [ ] L4 删除生产 hook 注册 + +## 验收标准 + +- [ ] fav / test / main 切模式失败 +- [ ] 每消费者独立 live 报告 +- [ ] 生产代码无 `OnOutbox =` / `AddOnApprovedHook(` + +## 相关链接 + +- Long-term: [.doc/ops/long-term/2026-09-15-event-shared-live-cutover.md](../long-term/2026-09-15-event-shared-live-cutover.md) +- Test report: (本波完成后补) + +## 证据 / 备注 diff --git a/cmd/eventmode/main.go b/cmd/eventmode/main.go new file mode 100644 index 00000000..ef6590d7 --- /dev/null +++ b/cmd/eventmode/main.go @@ -0,0 +1,113 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "time" + + "chat-agent-system/internal/config" + "chat-agent-system/internal/events" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + envPath := os.Getenv("EVENTMODE_ENV") + if envPath == "" { + envPath = "configs/.env.dev" + } + cfg, err := config.Load(envPath) + if err != nil { + fmt.Fprintln(os.Stderr, "eventmode: load env") + os.Exit(1) + } + if cfg.Env != events.OpsCutoverEnvironment { + fmt.Fprintln(os.Stderr, "eventmode: env must be dev") + os.Exit(1) + } + if cfg.ContextDbName != "agent_system_dev" { + fmt.Fprintln(os.Stderr, "eventmode: database must be agent_system_dev") + os.Exit(1) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + pool, err := pgxpool.New(ctx, cfg.ContextDSN()) + if err != nil { + fmt.Fprintln(os.Stderr, "eventmode: pool") + os.Exit(1) + } + defer pool.Close() + + switch os.Args[1] { + case "inventory": + if err := runInventory(ctx, pool); err != nil { + fmt.Fprintln(os.Stderr, "eventmode: inventory failed") + os.Exit(1) + } + case "set": + if len(os.Args) != 4 { + usage() + os.Exit(2) + } + consumer, mode := os.Args[2], os.Args[3] + before, _, _ := events.LoadMode(ctx, pool, cfg.Env, consumer) + if err := events.ApplyOpsCutover(ctx, pool, cfg.Env, consumer, mode); err != nil { + fmt.Fprintln(os.Stderr, err.Error()) + os.Exit(1) + } + after, rev, _ := events.LoadMode(ctx, pool, cfg.Env, consumer) + enc := json.NewEncoder(os.Stdout) + _ = enc.Encode(map[string]any{ + "environment": cfg.Env, "consumer": consumer, + "before": before, "after": after, "revision": rev, + }) + case "quarantine-illegal": + n, err := events.QuarantineIllegalPending(ctx, pool, "cursor-ops", "phase2-shared-live") + if err != nil { + fmt.Fprintln(os.Stderr, "eventmode: quarantine failed") + os.Exit(1) + } + _ = json.NewEncoder(os.Stdout).Encode(map[string]any{"quarantined": n}) + default: + usage() + os.Exit(2) + } +} + +func runInventory(ctx context.Context, pool *pgxpool.Pool) error { + modes, err := events.ListModes(ctx, pool) + if err != nil { + return err + } + backlog, err := events.ListBacklog(ctx, pool) + if err != nil { + return err + } + pending, err := events.ClassifyPending(ctx, pool) + if err != nil { + return err + } + type wire struct { + EventID string `json:"event_id"` + Consumer string `json:"consumer"` + Classification string `json:"classification"` + Reason string `json:"reason"` + } + class := make([]wire, 0, len(pending)) + for _, r := range pending { + class = append(class, wire{r.EventID, r.Consumer, r.Classification, r.Reason}) + } + return json.NewEncoder(os.Stdout).Encode(map[string]any{ + "modes": modes, "backlog": backlog, "pending_classifications": class, + }) +} + +func usage() { + fmt.Fprintln(os.Stderr, "usage: eventmode inventory | set | quarantine-illegal") +} diff --git a/internal/events/cutover.go b/internal/events/cutover.go new file mode 100644 index 00000000..c5078b8a --- /dev/null +++ b/internal/events/cutover.go @@ -0,0 +1,170 @@ +package events + +import ( + "context" + "fmt" + "strings" + + "chat-agent-system/internal/app/uow" + + "github.com/jackc/pgx/v5/pgxpool" +) + +const OpsCutoverEnvironment = "dev" + +// ValidateOpsCutover is the shared-dev gate for cluster mode changes. +// test/main, FAV, and unknown consumers are rejected before any write. +func ValidateOpsCutover(environment, consumer, mode string) error { + if strings.TrimSpace(environment) != OpsCutoverEnvironment { + return fmt.Errorf("%w: environment %q", ErrOpsCutover, environment) + } + switch consumer { + case ConsumerNotify, ConsumerWorkflowAdvance, ConsumerProjection: + default: + return fmt.Errorf("%w: consumer %q", ErrOpsCutover, consumer) + } + switch mode { + case ModeOff, ModeShadow, ModeLive: + default: + return fmt.Errorf("%w: mode %q", ErrOpsCutover, mode) + } + return nil +} + +func ApplyOpsCutover(ctx context.Context, pool *pgxpool.Pool, environment, consumer, mode string) error { + if err := ValidateOpsCutover(environment, consumer, mode); err != nil { + return err + } + return SetMode(ctx, pool, environment, consumer, mode) +} + +type ModeSnapshot struct { + Environment string `json:"environment"` + Consumer string `json:"consumer"` + Mode string `json:"mode"` + Revision int64 `json:"revision"` +} + +type BacklogRow struct { + Consumer string `json:"consumer"` + Status string `json:"status"` + Count int64 `json:"count"` +} + +func ListModes(ctx context.Context, pool *pgxpool.Pool) ([]ModeSnapshot, error) { + rows, err := pool.Query(ctx, ` + SELECT environment, consumer, mode, revision + FROM event_consumer_modes + ORDER BY environment, consumer + `) + if err != nil { + return nil, err + } + defer rows.Close() + var out []ModeSnapshot + for rows.Next() { + var m ModeSnapshot + if err := rows.Scan(&m.Environment, &m.Consumer, &m.Mode, &m.Revision); err != nil { + return nil, err + } + out = append(out, m) + } + return out, rows.Err() +} + +func ListBacklog(ctx context.Context, pool *pgxpool.Pool) ([]BacklogRow, error) { + rows, err := pool.Query(ctx, ` + SELECT consumer, status, COUNT(*) + FROM event_deliveries + GROUP BY consumer, status + ORDER BY consumer, status + `) + if err != nil { + return nil, err + } + defer rows.Close() + var out []BacklogRow + for rows.Next() { + var r BacklogRow + if err := rows.Scan(&r.Consumer, &r.Status, &r.Count); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +type pendingDelivery struct { + EventID string + Consumer string + EventName string + Environment string + ProjectID int64 + SchemaVersion int + PayloadDigest string +} + +func loadPending(ctx context.Context, pool *pgxpool.Pool) ([]pendingDelivery, error) { + rows, err := pool.Query(ctx, ` + SELECT d.event_id, d.consumer, e.event_name, e.environment, e.project_id, e.schema_version, COALESCE(e.payload_digest, '') + FROM event_deliveries d + INNER JOIN event_envelopes e ON e.event_id = d.event_id + WHERE d.status IN ('pending','leased') + `) + if err != nil { + return nil, err + } + defer rows.Close() + var out []pendingDelivery + for rows.Next() { + var p pendingDelivery + if err := rows.Scan(&p.EventID, &p.Consumer, &p.EventName, &p.Environment, &p.ProjectID, &p.SchemaVersion, &p.PayloadDigest); err != nil { + return nil, err + } + out = append(out, p) + } + return out, rows.Err() +} + +func ClassifyPending(ctx context.Context, pool *pgxpool.Pool) ([]InventoryRow, error) { + pending, err := loadPending(ctx, pool) + if err != nil { + return nil, err + } + out := make([]InventoryRow, 0, len(pending)) + for _, p := range pending { + env := Envelope{ + EventID: p.EventID, EventName: p.EventName, SchemaVersion: p.SchemaVersion, + Environment: p.Environment, ProjectID: p.ProjectID, PayloadDigest: p.PayloadDigest, + } + out = append(out, ClassifyEnvelope(env, p.Consumer)) + } + return out, nil +} + +// QuarantineIllegalPending persists quarantine for pending deliveries whose +// consumer is not on the contract (including FAV). Safe envelopes are left untouched. +func QuarantineIllegalPending(ctx context.Context, pool *pgxpool.Pool, operator, approval string) (int, error) { + if operator == "" || approval == "" { + return 0, ErrIntegrityFailure + } + rows, err := ClassifyPending(ctx, pool) + if err != nil { + return 0, err + } + n := 0 + unit := uow.NewPG(pool) + for _, row := range rows { + if row.Classification != ClassificationQuarantine { + continue + } + err := unit.Within(ctx, func(scope uow.TxScope) error { + return PersistInventory(uow.WithScope(ctx, scope), row, operator, approval) + }) + if err != nil { + return n, err + } + n++ + } + return n, nil +} diff --git a/internal/events/cutover_script_test.go b/internal/events/cutover_script_test.go new file mode 100644 index 00000000..f928a524 --- /dev/null +++ b/internal/events/cutover_script_test.go @@ -0,0 +1,44 @@ +package events_test + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestEventConsumerModeScriptGates(t *testing.T) { + root := repoRoot(t) + text, err := os.ReadFile(filepath.Join(root, "scripts", "event-consumer-mode.sh")) + if err != nil { + t.Fatal(err) + } + s := string(text) + if !strings.Contains(s, "fav") || !strings.Contains(s, "refusing test/main") { + t.Fatal("script must reject fav and test/main") + } + if strings.Contains(s, "5434") || strings.Contains(s, "5435") { + t.Fatal("script must not pin test/main ports") + } + if !strings.Contains(s, "agent_system_dev") || !strings.Contains(s, "5433") { + t.Fatal("script must pin shared dev db") + } +} + +func repoRoot(t *testing.T) string { + t.Helper() + dir, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir + } + parent := filepath.Dir(dir) + if parent == dir { + t.Fatal("go.mod not found") + } + dir = parent + } +} diff --git a/internal/events/cutover_test.go b/internal/events/cutover_test.go new file mode 100644 index 00000000..36658e37 --- /dev/null +++ b/internal/events/cutover_test.go @@ -0,0 +1,34 @@ +package events + +import "testing" + +func TestValidateOpsCutoverRejectsFavAndNonDev(t *testing.T) { + if err := ValidateOpsCutover("dev", ConsumerNotify, ModeShadow); err != nil { + t.Fatal(err) + } + if err := ValidateOpsCutover("test", ConsumerNotify, ModeLive); err == nil { + t.Fatal("test env must fail") + } + if err := ValidateOpsCutover("main", ConsumerNotify, ModeLive); err == nil { + t.Fatal("main env must fail") + } + if err := ValidateOpsCutover("dev", ConsumerFAV, ModeOff); err == nil { + t.Fatal("fav must fail even at off") + } + if err := ValidateOpsCutover("dev", "unknown", ModeLive); err == nil { + t.Fatal("unknown consumer must fail") + } + if err := ValidateOpsCutover("dev", ConsumerNotify, "dual"); err == nil { + t.Fatal("unknown mode must fail") + } +} + +func TestValidateOpsCutoverAllowsThreeConsumers(t *testing.T) { + for _, c := range []string{ConsumerNotify, ConsumerWorkflowAdvance, ConsumerProjection} { + for _, m := range []string{ModeOff, ModeShadow, ModeLive} { + if err := ValidateOpsCutover("dev", c, m); err != nil { + t.Fatalf("%s %s: %v", c, m, err) + } + } + } +} diff --git a/internal/events/envelope.go b/internal/events/envelope.go index ddd7bfa4..039c62e3 100644 --- a/internal/events/envelope.go +++ b/internal/events/envelope.go @@ -23,6 +23,7 @@ var ( ErrDisabledConsumer = errors.New("events: consumer is disabled") ErrUnregistered = errors.New("events: event or consumer is not in the static registry") ErrReplayApply = errors.New("events: replay apply rejected") + ErrOpsCutover = errors.New("events: shared cutover rejected") ) type Envelope struct { diff --git a/scripts/event-consumer-mode.sh b/scripts/event-consumer-mode.sh new file mode 100755 index 00000000..006fd865 --- /dev/null +++ b/scripts/event-consumer-mode.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash +# Shared agent-dev event consumer mode switch. +# Only environment=dev, consumers notify|workflow_advance|projection, modes off|shadow|live. +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT" + +ENV_FILE="${EVENTMODE_ENV:-configs/.env.dev}" + +usage() { + echo "usage: $0 inventory | set | quarantine-illegal" >&2 + exit 2 +} + +if [[ $# -lt 1 ]]; then + usage +fi + +if [[ ! -f "$ENV_FILE" ]]; then + echo "event-consumer-mode: missing $ENV_FILE" >&2 + exit 1 +fi + +if grep -E -q '^ENV=(test|main)[[:space:]]*$' "$ENV_FILE"; then + echo "event-consumer-mode: refusing test/main env file" >&2 + exit 1 +fi + +# Host-side ops pin the published dev database; never test/main ports. +export CONTEXT_DB_HOST="${CONTEXT_DB_HOST:-127.0.0.1}" +export CONTEXT_DB_PORT="${CONTEXT_DB_PORT:-5433}" +export CONTEXT_DB_NAME="${CONTEXT_DB_NAME:-agent_system_dev}" +export ENV=dev +export EVENTMODE_ENV="$ENV_FILE" + +case "$1" in + inventory) + go run ./cmd/eventmode inventory + ;; + set) + if [[ $# -ne 3 ]]; then + usage + fi + consumer="$2" + mode="$3" + if [[ "$consumer" == "fav" ]]; then + echo "event-consumer-mode: fav is disabled" >&2 + exit 1 + fi + go run ./cmd/eventmode set "$consumer" "$mode" + ;; + quarantine-illegal) + go run ./cmd/eventmode quarantine-illegal + ;; + *) + usage + ;; +esac