feat(events): gate shared-dev consumer mode changes
Add ValidateOpsCutover, eventmode CLI, and a host script that pin agent_system_dev and refuse FAV plus test/main before any cluster mode write. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
+3
-2
@@ -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/<slug>` 或 `fix/<slug>`，**禁止** `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`
|
||||
|
||||
|
||||
@@ -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: （本波完成后补）
|
||||
|
||||
## 证据 / 备注
|
||||
@@ -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 <consumer> <mode> | quarantine-illegal")
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Executable
+59
@@ -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 <consumer> <mode> | 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
|
||||
Reference in New Issue
Block a user