diff --git a/.task/feature-rdm-two-trees-project-audit.md b/.task/feature-rdm-two-trees-project-audit.md new file mode 100644 index 0000000..04598fb --- /dev/null +++ b/.task/feature-rdm-two-trees-project-audit.md @@ -0,0 +1,39 @@ +# W9 Backend project audit candidate + +- Scope: TT-A01/A02/A04/U04, approved stage 5 W9. Stage 6 local implementation; human review, formal acceptance, PR, CI, tag, and deployment are pending. +- Branch/worktree: `feature/rdm-two-trees-project-audit` at `/home/weichen/projects/Agent-System-Backend-wt-feature-rdm-two-trees-project-audit`. +- R0: Backend authoritative `origin/main` `5d5fc9891c0eb8994545ccb8b740535d9b99fb0e`; main was clean and equal to the fetched remote before this worktree was created. Core gitlink in this worktree is `5773e74086791731059730bd3726dc8c9cd1381f`. +- Startup receipt: `/tmp/backend-w9-precheck-20261014-01.json` (SELF, exit 2, `SYNC_REQUIRED` for the separate Dev checkout `feature/rdm-increment-acceptance-note`, six commits behind Dev `origin/main`). This Backend task uses the independent-development path in Backend `AGENTS.md`; it does not write that Dev checkout or its task card. The receipt is preserved as a failure, not relabeled READY. +- Approved affected design: `rdm-two-trees-stage6-audit-link-design-adjustment-r1.md`, SHA256 `385d9d7570217ddcce9fe60a60c99dc58c9a4875ad790584a724d09a210a6e52`; user approval in the task session covers this local Backend increment, not tag publication or Stage 7 acceptance. +- Approved PostgreSQL cutoff adjustment: `rdm-two-trees-stage6-w9-pg-audit-cutoff-adjustment-r1.md`, SHA256 `2357b715a7c8f3451b34cc25a578d655e8a48f3f26d9cf7561ea7a13998ca857`; scope is isolated Backend implementation/testing, with no real database migration or release authorization. +- Approved all-writers affected design: `rdm-two-trees-stage6-w9-all-writers-cutoff-adjustment-r1.md`, SHA256 `44f6f7cd34bf95a9c1eb844bfdd70a5c2233e1b5ffc89b3091359dcb5fc3b5d8`; the direct user approval allows the local v34 correction and isolated tests only. + +## Implemented local slice + +- Optional `ProjectAuditHost` and domain `audit.read` registration. A different plugin, absent registration, or unbound actor cannot read. +- Store write snapshots the project from the owner-matched conversation; query requires the authenticated actor's owner and that immutable project. This deliberately excludes other members' private conversations and unbound historical rows, and does not move old receipts when a conversation changes projects. +- Command audit stores the current verified account display name at write time; old rows expose `not-recorded`. Public receipts omit raw payload and parameter digest. Cursor uses project-and-owner scoped source cutoff. +- Initial commit `c8dd169a1fdc1451fab3b475c14f9636a7f5bf6c` exposed only the actor-owned project slice. The next local increment fixes button actor propagation and adds a trusted domain-commit link publisher, verifier callback, exact linked metadata read, current `audit.read` and per-object metadata checks, and durable idempotency. Unlinked personal conversations remain private. +- Trusted command dispatch now creates an operation ID before calling the compiled handler and passes that ID through `extension.Request`. The handler can commit the same ID with its domain event/outbox; the final Backend audit/receipt uses it. Client `params.operationId` is ignored. An outbox link request with `HostAuditOffset=0` resolves a unique committed host row by owner, fixed project, and operation ID; a missing row returns retryable `ErrProjectAuditPending`, while duplicate rows fail closed. The separate idempotency key still controls retry equivalence. +- Independent review of `a8fa2fe` found two P2 security defects: clock-based operation IDs could collide; and `NewScopedHost(owner,"rdm")` let another compiled plugin self-report the publisher's plugin ID. The new local increment uses 192 bits from `crypto/rand` for each host-generated operation ID. Both zero-offset and explicit-offset publication require a unique committed row for owner/project/operation; a historical duplicate rejects publication, including attempts to link another private command. Verifier registration returns a private `ProjectAuditPublisherFactory` for the registering plugin's background outbox; a generic scoped Host has no publisher method. Fixed-clock concurrent IDs, forged params, ambiguous private offset, forged scoped Host, and legitimate registered publisher each have focused tests. +- The minted publisher exposes only `ProjectAuditPublisher`, not the full owner-scoped `Host`; tests reject attempts to use `NewScopedHost` with a self-reported `rdm` string as publisher authority. A publisher is a registration-time capability, while `IdempotencyKey` remains the separate retry identity. +- Exact-source linked reads do not enumerate hidden links or reveal raw payload, private conversation ID, or content. Body disclosure and a broad project link list are not implemented; the public coverage reports the narrower capability. +- A PostgreSQL-only v33 default function takes a table-specific transaction advisory lock before assigning `command_audit.id`, making supported implicit-ID writes commit in source-offset order. Migration and each PG append/project page verify CACHE 1, ascending non-cyclic positive IDs, the exact guarded default expression, and the next ID above the committed maximum; unsafe drift fails closed. SQLite remains single-writer. The owner-only project page reports `committed-prefix` only after runtime checks. v34 below closes the explicit-ID bypass; online sequence maintenance remains prohibited. +- Isolated PG RED receipt `/tmp/backend-w9-explicit-low-red-20261014-01.json` proved v33 still misreported `committed-prefix`: explicit ID 3 committed after normal ID 4 and disappeared across pages. Local v34 adds `ENABLE ALWAYS` insert and append-only mutation triggers. Every online INSERT checks READ COMMITTED, obtains the same transaction lock, and rejects a nonpositive, nonmonotone, or beyond-reserved ID. UPDATE/DELETE reject. PG append and every strong project page verify both trigger identities, exact event shape, and enablement; the existing default allocator also checks this guard. This is a local candidate, not yet independently reviewed or deployed. +- `storage-migrate` now copies `command_audit` within one ACCESS EXCLUSIVE transaction: it disables only the named insert guard, copies explicit IDs, verifies count and checksum, advances the owned sequence only forward, restores and verifies the guard, and records done before commit. Failure rolls back table data, trigger state, and done; nontransactional `setval` may leave a safe forward gap. Idempotent done checks guard health before reporting success. Other tables retain their existing path. + +## Verification and remaining work + +- The session saw RED for the missing store and host interfaces, unscoped cutoff, cross-member private disclosure, and conversation-project reassignment before the respective fixes. These RED outputs were not captured by `make evidence`, so this card does not use them as formal evidence. +- SELF receipt `/tmp/backend-w9-go-test-20261014-01.json`: `go test ./...`, exit 0, run on the pre-commit worktree based on R0. +- SELF receipt `/tmp/backend-w9-build-20261014-01.json`: `CGO_ENABLED=0 go build ./cmd/agent-system`, exit 0, run on that pre-commit tree. Generated local binary removed after the build. +- Historical SELF PostgreSQL receipt `/tmp/backend-w9-pg-cutoff-20261014-01.json`: isolated PostgreSQL 18.6 instance on port 55442 reproduced a low-ID transaction committing after a higher visible ID. `/tmp/backend-w9-pg-link-20261014-02.json` verifies the link table, retry/conflict behavior, and the pre-v33 cutoff counterexample. These receipts describe the old `offset-upper-bound-only` candidate, not the new v33 guarantee. +- SELF PostgreSQL receipt `/tmp/backend-w9-pg-migrate-20261014-01.json`: isolated SQLite→PostgreSQL migration preserved one linked receipt and advanced its target sequence; the same migration ID replayed idempotently. The temporary PG instance was stopped and removed after verification. +- SELF PostgreSQL receipt `/tmp/backend-w9-pg-operation-20261014-01.json`: a fresh isolated PG instance reran the link lookup, idempotency/conflict, and low-ID late-commit checks after trusted operation-offset resolution was added. The instance was stopped and removed. +- New PG cross-page RED receipt `/tmp/backend-w9-pg-crosspage-red-20261014-01.json`: before v33, page 1 returned offsets 4 and 2, page 2 returned 1 after offset 3 committed, and the paged view omitted 3. The same test now passes with v33. Isolated PG tests also cover CACHE 10 counterexample/rejection, CACHE 1, sequence rewind into a vacant low ID, down/up with unchanged link IDs, advisory-lock timeout and rollback gaps, and SQLite→PG explicit-ID migration. +- Historical pre-commit SELF load receipt `/tmp/backend-w9-pg-load-precommit-20261014-01.json` captured an initial v33 variant whose unlocked status function later produced false `cutoff unsafe` errors under concurrency; it is **not** valid performance acceptance evidence. The corrected status check reads committed MAX before non-MVCC sequence state and does not lock readers; the default ID allocator alone holds the transaction advisory lock and rechecks after `nextval`. `TestProjectAuditConcurrentReadWriteNeverReportsFalseDrift` and repeated PG load/cross-page runs protect that ordering. Three-condition pre-commit SELF raw receipts: `/tmp/backend-w9-pg-load-readlock-precommit-20261014-01.json` uses a test-only locked status wrapper, while `/tmp/backend-w9-pg-load-unlocked2-precommit-20261014-01.json` uses the final MAX-first unlocked status; each run also logs its v32 direct-INSERT baseline, PostgreSQL 18.6 version, 192 writes per mode/8 workers, elapsed time, p95, and throughput. The wrapper adds overhead and all measurements vary with hardware/load; no performance threshold has been approved, so they are Stage 6 cost evidence only. +- Pre-commit SELF `/tmp/backend-w9-default-drift-red-20261014-01.json` showed that a replacement default containing the guarded function name but executing unguarded `nextval` was accepted; `/tmp/backend-w9-default-drift-green-20261014-01.json` verifies exact-expression rejection. Pre-commit SELF `/tmp/backend-w9-precommit-all-20261014-02.json` and `/tmp/backend-w9-precommit-pg-20261014-02.json` passed before this last guard tightening; fixed-HEAD runs are required below. +- After the guard tightening, pre-commit SELF receipts `/tmp/backend-w9-precommit-all-final-20261014-01.json` (`go test ./... -count=1`), `/tmp/backend-w9-precommit-pg-final-20261014-01.json` (isolated PG store/storage-migrate), and `/tmp/backend-w9-precommit-pg-race-20261014-01.json` (cross-page/concurrent read-write with race detector) all exited 0. These still identify the dirty pre-commit tree, so they do not replace fixed-HEAD verification. +- v34 pre-commit SELF `/tmp/backend-w9-allwriters-load-precommit-20261014-01.json` records isolated PostgreSQL 18.6, three runs of 192 writes per mode with 8 workers, comparing v32 direct INSERT with v34 guarded `Store.AppendAudit`; it also logs three mixed runs of 180 writes with 6 writers and 160 200-row pages with 4 readers. Protected write throughput ranged about 2.7–3.0k/s, write p95 6.4–6.6ms, and mixed large-page p95 1.2–1.9ms on this one host. These are SELF cost signals with no approved threshold and no Stage 7 acceptance meaning; fixed-HEAD tests still follow. +- Integration constraint: this branch uses schema v31 (audit attribution), v32 (links), v33 (PG implicit cutoff), and v34 (all-writer guard), while the independent W6 Backend branch also has a local v31. Neither branch has consumed the other's unmerged work. After the first branch lands, the later branch owner must sync the new main, renumber its unpublished migrations without rewriting an already released version, and rerun fresh/old SQLite and PG migration plus combined regression on the actual merged source. These isolated tests do not prove W6+W9 compatibility. +- Required next W9 work: real RDM committed-event verifier/outbox consumer, Backend project-event source, run ID propagation, model/tool/background receipts, body/detail disclosure, broad authorized linked-operation pagination, history/export, and combined product verification remain unfinished. A domain commit followed by a lost Backend audit append leaves the link pending; recovery must retain and reconcile that unknown state without claiming a host receipt. Exact-link and owner-only paging do not satisfy Stage 7 audit acceptance. PostgreSQL v33 adds a global command-audit append lock; its measured wait/throughput cost has no approved performance threshold and needs human Stage 6 review. diff --git a/cmd/storage-migrate/integration_test.go b/cmd/storage-migrate/integration_test.go index 5fa87c9..87b05d9 100644 --- a/cmd/storage-migrate/integration_test.go +++ b/cmd/storage-migrate/integration_test.go @@ -128,6 +128,15 @@ func TestMigrateEndToEndAndIdempotent(t *testing.T) { if _, err := db.Exec(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner) VALUES ('2026-10-11T03:00:00Z','cmd-2','http','chat','done','op-2','c1','{}','alice')`); err != nil { t.Fatalf("序列推进失败（复制后插入撞主键）：%v", err) } + var linkedResource string + if err := db.QueryRow(`SELECT resource_id FROM project_audit_links WHERE host_audit_offset=7 AND source_id='commit-1'`).Scan(&linkedResource); err != nil || linkedResource != "doc-1" { + t.Fatalf("审计链接迁移丢失: resource=%q err=%v", linkedResource, err) + } + var newLinkID int64 + if err := db.QueryRow(`INSERT INTO project_audit_links(host_audit_offset,project_id,owner,source_kind,source_id,operation_id,resource_kind,resource_id,run_id,disclosure_class,idempotency_key,created_at) + VALUES (7,'p1','alice','domain-commit','commit-2','op-1','document','doc-2','','metadata','outbox-2','2026-10-11T03:00:01Z') RETURNING id`).Scan(&newLinkID); err != nil || newLinkID <= 3 { + t.Fatalf("审计链接序列未推进: id=%d err=%v", newLinkID, err) + } // 报告不含秘密值。 var report string if err := db.QueryRow(`SELECT report FROM migration_control WHERE migration_id='mig-e2e'`).Scan(&report); err != nil || strings.Contains(report, "sk-secret-value") { @@ -135,6 +144,105 @@ func TestMigrateEndToEndAndIdempotent(t *testing.T) { } } +func TestCommandAuditExclusiveCopyRollsBackAndNeverRewindsSequence(t *testing.T) { + target, cleanup := pgSchemaDSN(t) + defer cleanup() + fixture := buildFixture(t) + src, err := openSourceRO(fixture) + if err != nil { + t.Fatal(err) + } + defer src.Close() + st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + db := st.DB() + if _, err := db.Exec(controlDDL); err != nil { + t.Fatal(err) + } + cols, pk, err := targetColumns(db, "command_audit") + if err != nil { + t.Fatal(err) + } + bools, err := boolColumns(db, "command_audit", cols) + if err != nil { + t.Fatal(err) + } + count, checksum, err := sourceScan(src, "command_audit", cols, pk, bools) + if err != nil { + t.Fatal(err) + } + blocker, err := db.Begin() + if err != nil { + t.Fatal(err) + } + var existing int + if err := blocker.QueryRow(`SELECT COUNT(*) FROM command_audit`).Scan(&existing); err != nil { + t.Fatal(err) + } + timeoutDB, err := sql.Open("pgx", store.AppendPGRuntimeParam(target, "lock_timeout", "100")) + if err != nil { + t.Fatal(err) + } + if _, err := migrateCommandAudit(timeoutDB, src, "retry-audit", cols, pk, bools, count, checksum); err == nil || !strings.Contains(err.Error(), "lock timeout") { + t.Fatalf("exclusive copy did not fail on a held reader: %v", err) + } + timeoutDB.Close() + if err := blocker.Rollback(); err != nil { + t.Fatal(err) + } + if _, err := migrateCommandAudit(db, src, "retry-audit", cols, pk, bools, count+1, checksum); err == nil { + t.Fatal("mismatched copy committed despite in-transaction target count check") + } + var rows, done int + if err := db.QueryRow(`SELECT COUNT(*) FROM command_audit`).Scan(&rows); err != nil || rows != 0 { + t.Fatalf("failed copy left rows: %d %v", rows, err) + } + if err := db.QueryRow(`SELECT COUNT(*) FROM migration_table_state WHERE table_name='command_audit'`).Scan(&done); err != nil || done != 0 { + t.Fatalf("failed copy recorded done: %d %v", done, err) + } + var enabled string + if err := db.QueryRow(`SELECT tgenabled FROM pg_trigger WHERE tgrelid='command_audit'::regclass AND tgname='project_audit_insert_guard_v1'`).Scan(&enabled); err != nil || enabled != "A" { + t.Fatalf("failed copy left guard disabled: %q %v", enabled, err) + } + var seq string + if err := db.QueryRow(`SELECT pg_get_serial_sequence('command_audit','id')`).Scan(&seq); err != nil { + t.Fatal(err) + } + // setval is non-transactional; an interrupted offline attempt can leave a + // forward gap even though its rows and trigger-state changes roll back. + tx, err := db.Begin() + if err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(`LOCK TABLE command_audit IN ACCESS EXCLUSIVE MODE`); err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(`ALTER TABLE command_audit DISABLE TRIGGER project_audit_insert_guard_v1`); err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(`SELECT setval($1::regclass,50,true)`, seq); err != nil { + t.Fatal(err) + } + if err := tx.Rollback(); err != nil { + t.Fatal(err) + } + if _, err := migrateCommandAudit(db, src, "retry-audit", cols, pk, bools, count, checksum); err != nil { + t.Fatalf("retry after rollback: %v", err) + } + var last int64 + if err := db.QueryRow(`SELECT last_value FROM ` + seq).Scan(&last); err != nil || last < 50 { + t.Fatalf("copy rewound sequence: last=%d err=%v", last, err) + } + var nextID int64 + if err := db.QueryRow(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload) + VALUES ('2026-10-14T00:00:00Z','after-copy','','','committed','','','alice','{}') RETURNING id`).Scan(&nextID); err != nil || nextID <= 50 { + t.Fatalf("post-copy ID did not advance: id=%d err=%v", nextID, err) + } +} + func TestMigrateResumeAfterInterrupt(t *testing.T) { target, cleanup := pgSchemaDSN(t) defer cleanup() diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index e3f5eda..63a6beb 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -287,7 +287,7 @@ var copyOrder = []string{ "projects", "users", "profiles", "settings", "plugins", "conversations", "runs", "messages", "events", "artifacts", "hitl_cards", "plans", "todos", "knowledge", "knowledge_vectors", - "activity_cards", "project_event_shares", "command_audit", "prompt_modules", + "activity_cards", "project_event_shares", "command_audit", "project_audit_links", "prompt_modules", "sessions", "api_keys", "plugin_grants", "plugin_schema", "plugin_kv", "background_tasks", "work_task_bindings", "memory_entries", "audit_events", "context_manifests", "workflow_step_sources", "pending_confirms", "llm_usage_daily", @@ -546,9 +546,18 @@ func migrateTable(db *sql.DB, src *sql.DB, migrationID, table string) (map[strin if countNow != srcCount || sumNow != srcChecksum { return nil, fmt.Errorf("源与登记不符（疑似换快照）：count %d/%d checksum %s…/%s…", countNow, srcCount, sumNow[:12], srcChecksum[:12]) } + if table == "command_audit" { + var guarded bool + if err := db.QueryRow(`SELECT project_audit_cutoff_safe('command_audit'::regclass) AND project_audit_writers_guarded('command_audit'::regclass)`).Scan(&guarded); err != nil || !guarded { + return nil, fmt.Errorf("command_audit idempotent guard is unsafe: safe=%v err=%v", guarded, err) + } + } entry := map[string]any{"table": table, "source_count": countNow, "target_count": dstCountStored, "checksum": sumNow, "state": "done(idempotent)"} return entry, nil } + if table == "command_audit" { + return migrateCommandAudit(db, src, migrationID, cols, pk, bools, countNow, sumNow) + } if err := copyRows(db, src, table, cols); err != nil { return nil, err } @@ -701,7 +710,11 @@ func sourceScan(src *sql.DB, table string, cols, pk []string, bools map[string]b return scanChecksum(rows, cols, bools) } -func targetScan(db *sql.DB, table string, cols, pk []string, bools map[string]bool) (int64, string, error) { +type rowsQuerier interface { + Query(string, ...any) (*sql.Rows, error) +} + +func targetScan(db rowsQuerier, table string, cols, pk []string, bools map[string]bool) (int64, string, error) { rows, err := db.Query(`SELECT ` + quoteJoin(cols, `"`) + ` FROM "` + table + `" ORDER BY ` + quoteJoin(pk, `"`)) if err != nil { return 0, "", err @@ -724,6 +737,18 @@ func copyRows(db *sql.DB, src *sql.DB, table string, cols []string) error { if err != nil { return err } + tx, err := db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if err := copyRowsInTx(tx, src, table, cols, bools); err != nil { + return err + } + return tx.Commit() +} + +func copyRowsInTx(tx *sql.Tx, src *sql.DB, table string, cols []string, bools map[string]bool) error { colList := quoteJoin(cols, `"`) ph := make([]string, len(cols)) for i := range ph { @@ -740,11 +765,6 @@ func copyRows(db *sql.DB, src *sql.DB, table string, cols []string) error { for i := range vals { ptrs[i] = &vals[i] } - tx, err := db.Begin() - if err != nil { - return err - } - defer tx.Rollback() stmt, err := tx.Prepare(fmt.Sprintf(`INSERT INTO "%s" (%s) VALUES (%s) ON CONFLICT DO NOTHING`, table, colList, placeholders)) if err != nil { return err @@ -772,7 +792,76 @@ func copyRows(db *sql.DB, src *sql.DB, table string, cols []string) error { if err := srcRows.Err(); err != nil { return err } - return tx.Commit() + return nil +} + +// command_audit alone needs an offline transaction: explicit source IDs do not +// pass the online monotone guard, and disabling it must never become visible +// separately from the copied rows, sequence advance, or done marker. +func migrateCommandAudit(db *sql.DB, src *sql.DB, migrationID string, cols, pk []string, bools map[string]bool, sourceCount int64, sourceChecksum string) (map[string]any, error) { + tx, err := db.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + if _, err := tx.Exec(`LOCK TABLE command_audit IN ACCESS EXCLUSIVE MODE`); err != nil { + return nil, err + } + if _, err := tx.Exec(`ALTER TABLE command_audit DISABLE TRIGGER project_audit_insert_guard_v1`); err != nil { + return nil, err + } + if err := copyRowsInTx(tx, src, "command_audit", cols, bools); err != nil { + return nil, err + } + targetCount, targetChecksum, err := targetScan(tx, "command_audit", cols, pk, bools) + if err != nil { + return nil, err + } + if targetCount != sourceCount || targetChecksum != sourceChecksum { + return nil, fmt.Errorf("command_audit copy mismatch: count %d/%d checksum %s/%s", targetCount, sourceCount, targetChecksum, sourceChecksum) + } + if err := advanceCommandAuditSequence(tx); err != nil { + return nil, err + } + if _, err := tx.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_insert_guard_v1`); err != nil { + return nil, err + } + var guarded bool + if err := tx.QueryRow(`SELECT project_audit_cutoff_safe('command_audit'::regclass) AND project_audit_writers_guarded('command_audit'::regclass)`).Scan(&guarded); err != nil || !guarded { + return nil, fmt.Errorf("command_audit guard not restored: safe=%v err=%v", guarded, err) + } + if _, err := tx.Exec(`INSERT INTO migration_table_state (migration_id, table_name, source_count, target_count, checksum, state, finished_at) + VALUES ($1,'command_audit',$2,$3,$4,'done',$5) + ON CONFLICT(migration_id, table_name) DO UPDATE SET source_count=excluded.source_count, target_count=excluded.target_count, checksum=excluded.checksum, state=excluded.state, finished_at=excluded.finished_at`, + migrationID, sourceCount, targetCount, sourceChecksum, nowISO()); err != nil { + return nil, err + } + if err := tx.Commit(); err != nil { + return nil, err + } + return map[string]any{"table": "command_audit", "source_count": sourceCount, "target_count": targetCount, "checksum": sourceChecksum, "state": "done"}, nil +} + +func advanceCommandAuditSequence(tx *sql.Tx) error { + var seq string + if err := tx.QueryRow(`SELECT pg_get_serial_sequence('command_audit','id')`).Scan(&seq); err != nil || seq == "" { + return fmt.Errorf("command_audit sequence missing: %v", err) + } + var last int64 + var called bool + if err := tx.QueryRow(fmt.Sprintf(`SELECT last_value,is_called FROM %s`, seq)).Scan(&last, &called); err != nil { + return err + } + var maxID sql.NullInt64 + if err := tx.QueryRow(`SELECT MAX(id) FROM command_audit`).Scan(&maxID); err != nil { + return err + } + if maxID.Valid && (maxID.Int64 > last || (maxID.Int64 == last && !called)) { + if _, err := tx.Exec(`SELECT setval($1::regclass,$2,true)`, seq, maxID.Int64); err != nil { + return err + } + } + return nil } func sqliteBool(v any) bool { @@ -790,30 +879,34 @@ func sqliteBool(v any) bool { // 后续应用插入会撞主键）。无序列表为空操作。 func advanceSequences(db *sql.DB, table string) error { rows, err := db.Query(` - SELECT pg_get_serial_sequence($1, column_name) + SELECT column_name, pg_get_serial_sequence(format('%I.%I',table_schema,table_name), column_name) FROM information_schema.columns - WHERE table_schema=current_schema() AND table_name=$1 - AND (column_default LIKE 'nextval%' OR is_identity='YES')`, table) + WHERE table_schema=current_schema() AND table_name=$1`, table) if err != nil { return err } defer rows.Close() - var seqs []string + var seqs []struct{ name, column string } for rows.Next() { - var s any - if err := rows.Scan(&s); err != nil { + var column string + var seq sql.NullString + if err := rows.Scan(&column, &seq); err != nil { return err } - if s != nil { - seqs = append(seqs, s.(string)) + if seq.Valid { + seqs = append(seqs, struct{ name, column string }{seq.String, column}) } } if err := rows.Err(); err != nil { return err } for _, s := range seqs { - if _, err := db.Exec(fmt.Sprintf(`SELECT setval('%s', COALESCE((SELECT MAX(id) FROM "%s"), 1))`, s, table)); err != nil { - return fmt.Errorf("setval %s: %w", s, err) + // Sequence ownership remains authoritative when a column's default is a + // guarded function rather than a literal nextval expression. + column := `"` + strings.ReplaceAll(s.column, `"`, `""`) + `"` + tableName := `"` + strings.ReplaceAll(table, `"`, `""`) + `"` + if _, err := db.Exec(fmt.Sprintf(`SELECT setval($1::regclass, COALESCE((SELECT MAX(%s) FROM %s), 1))`, column, tableName), s.name); err != nil { + return fmt.Errorf("setval %s (%s): %w", s.name, s.column, err) } } return nil @@ -822,7 +915,7 @@ func advanceSequences(db *sql.DB, table string) error { // checkOwners：业务行引用的 owner 必须存在于 users（单租户默认 ”/'local' // 除外）。未知 owner 阻塞迁移（U4/R7：不静默带过未知归属）。 func checkOwners(db *sql.DB, src *sql.DB) error { - ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "project_event_shares", "command_audit", "work_task_bindings", "context_manifests", "workflow_step_sources"} + ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "project_event_shares", "command_audit", "project_audit_links", "work_task_bindings", "context_manifests", "workflow_step_sources"} for _, t := range ownerTables { if !sqliteHasTable(src, t) { continue diff --git a/cmd/storage-migrate/main_test.go b/cmd/storage-migrate/main_test.go index 9e60ae3..e89ee3f 100644 --- a/cmd/storage-migrate/main_test.go +++ b/cmd/storage-migrate/main_test.go @@ -73,8 +73,10 @@ func seedEdgeRows(t *testing.T, st *store.Store) { mustExec(t, st, `INSERT INTO activity_cards (id, project_id, conversation_id, ref, owner, payload, status, pinned, created_at, updated_at, version) VALUES ('ac1', 'p1', 'c1', '', 'alice', '{"title":"卡"}', 'active', 1, '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 2)`) // 命令审计（SERIAL 主键：显式 id 插入）。 - mustExec(t, st, `INSERT INTO command_audit (id, time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner) VALUES - (7, '2026-10-11T01:00:00Z', 'cmd-1', 'http', 'chat', 'done', 'op-1', 'c1', '{"actor":"alice"}', 'alice')`) + mustExec(t, st, `INSERT INTO command_audit (id, time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner, project_id, actor_name) VALUES + (7, '2026-10-11T01:00:00Z', 'cmd-1', 'http', 'chat', 'committed', 'op-1', 'c1', '{"actor":"alice","commandId":"cmd-1","operationId":"op-1","state":"committed","conversationId":"c1"}', 'alice', 'p1', '爱丽丝')`) + mustExec(t, st, `INSERT INTO project_audit_links (id,host_audit_offset,project_id,owner,source_kind,source_id,operation_id,resource_kind,resource_id,run_id,disclosure_class,idempotency_key,created_at) VALUES + (3,7,'p1','alice','domain-commit','commit-1','op-1','document','doc-1','','metadata','outbox-1','2026-10-11T01:00:01Z')`) // 设置（'' 默认租户 scope）。 mustExec(t, st, `INSERT INTO settings (owner, scope, "values") VALUES ('', 'runtime', '{"rateLimitPerMin":0}')`) // 秘密字段（值不外泄：迁移按行复制，报告无正文）。 diff --git a/internal/app/app.go b/internal/app/app.go index 9be10b5..0bac7bd 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -559,17 +559,7 @@ func Run(opt Options) error { // U5：页面动作分发口——surfacesvc.SubmitAction 经命令总线四入口同源分发 //（审计信封、参数校验、DangerHigh 内联确认卡全复用；Confirmed 不透传）。 surfaceSvc.SetDispatcher(func(ctx context.Context, d surfacesvc.ActionDispatch) (surfacesvc.ActionOutcome, error) { - resp := bus.Dispatch(ctx, cmdbus.Request{ID: d.CommandID, Params: d.Params, - Entrance: "button", Owner: d.Owner, ConversationID: d.ConversationID}) - state := "committed" - if resp.Receipt.State != "" { - state = string(resp.Receipt.State) - } - cardID := "" - if resp.Card != nil { - cardID = string(resp.Card.ID) - } - return surfacesvc.ActionOutcome{State: state, CardID: cardID, Receipt: resp.Receipt, Payload: resp.Payload}, nil + return dispatchSurfaceAction(ctx, bus, d) }) // U3：generative-ui 工具注入闸——activation 台账关闭时命令不投影给该 owner // 的模型（禁用即不注入能力；执行端守卫仍在 surfacesvc）。 @@ -600,7 +590,7 @@ func Run(opt Options) error { authzSlot := &projectAuthz{} projectSlot := &projectLifecycleSlot{} contextSlot := &projectContextSlot{} - caps := hostcaps{projects: projectSlot, contexts: contextSlot, st: st, pipe: pipe, set: setSvc, authz: authzSlot, surfaces: surfaceSvc} + caps := hostcaps{projects: projectSlot, contexts: contextSlot, projectAudit: &projectAuditSlot{}, st: st, pipe: pipe, set: setSvc, authz: authzSlot, surfaces: surfaceSvc} // EXT-13：回合内检索（knowledge_search / @knowledge 注入）与 HTTP/Host 检索面 // 同一适配器授权；槽为共享指针，扩展在 applyCompiled 期间注册后即生效。 mgr.SetSharedKnowledgeAllow(func(owner string, projectID core.ID) bool { diff --git a/internal/app/compiled.go b/internal/app/compiled.go index f73b6f3..faf73d4 100644 --- a/internal/app/compiled.go +++ b/internal/app/compiled.go @@ -68,7 +68,7 @@ func (r pluginRegistry) AddCommandHandler(spec json.RawMessage, h extension.Hand conversationID: core.ID(req.ConversationID), hostProjectID: core.ID(req.HostProjectID)} return h(ctx, host, extension.Request{ ID: req.ID, Params: req.Params, Entrance: req.Entrance, - Actor: req.Actor, Owner: req.Owner, ConversationID: req.ConversationID, + Actor: req.Actor, Owner: req.Owner, ConversationID: req.ConversationID, OperationID: req.OperationID, HostProjectID: req.HostProjectID, }) }) diff --git a/internal/app/hostcaps.go b/internal/app/hostcaps.go index 40025bb..2afc768 100644 --- a/internal/app/hostcaps.go +++ b/internal/app/hostcaps.go @@ -21,13 +21,14 @@ import ( // 同一实现同时服务 hostapi.Server 的 Search/Complete 闭包与进程内 // inprocHost（EXT-1 批注 1：复用现有 owner 作用域函数，不新造句柄）。 type hostcaps struct { - projects *projectLifecycleSlot - contexts *projectContextSlot - st *store.Store - pipe *rag.Pipeline - set *settings.Service - authz *projectAuthz // EXT-8 适配器槽（共享指针；首个注册者生效） - surfaces *surfacesvc.Service // EXT-17 Surface 能力槽（nil = 未装配，明确拒绝） + projects *projectLifecycleSlot + contexts *projectContextSlot + projectAudit *projectAuditSlot + st *store.Store + pipe *rag.Pipeline + set *settings.Service + authz *projectAuthz // EXT-8 适配器槽（共享指针；首个注册者生效） + surfaces *surfacesvc.Service // EXT-17 Surface 能力槽（nil = 未装配，明确拒绝） } // projectAuthz EXT-8 资源授权适配器槽：进程内扩展注册，组合根装配进 Server。 diff --git a/internal/app/project_audit.go b/internal/app/project_audit.go new file mode 100644 index 0000000..43a7183 --- /dev/null +++ b/internal/app/project_audit.go @@ -0,0 +1,336 @@ +package app + +import ( + "context" + "errors" + "sync" + + "agent-system-backend/internal/store" + "agent-system-backend/pkg/extension" + core "agent-system-core" +) + +// projectAuditSlot binds the audit.read policy to its registering domain plugin. +// A different plugin cannot borrow that policy through its own Host handle. +type projectAuditSlot struct { + mu sync.RWMutex + holder string + check extension.ProjectAuditAuthorizer + verifier extension.ProjectAuditLinkVerifier + disclose extension.ProjectAuditDisclosureAuthorizer + publisherToken *auditPublisherToken +} + +// This identity is available only to the publisher factory returned at +// registration. A caller-provided pluginID cannot manufacture the pointer. +type auditPublisherToken struct{ marker byte } + +type projectAuditPublisherFactory struct { + caps hostcaps + pluginID string + token *auditPublisherToken +} + +func (f projectAuditPublisherFactory) ForOwner(owner string) (extension.ProjectAuditPublisher, error) { + if owner == "" || f.token == nil { + return nil, extension.ErrProjectAuditDenied + } + return projectAuditPublisher{caps: f.caps, owner: owner, pluginID: f.pluginID, token: f.token}, nil +} + +type projectAuditPublisher struct { + caps hostcaps + owner string + pluginID string + token *auditPublisherToken +} + +func (p projectAuditPublisher) PublishProjectAuditLink(ctx context.Context, req extension.ProjectAuditLinkRequest) (extension.ProjectAuditLinkReceipt, error) { + h := inprocHost{caps: p.caps, owner: p.owner, pluginID: p.pluginID} + return h.publishProjectAuditLink(ctx, req, p.token) +} + +func (s *projectAuditSlot) register(pluginID string, check extension.ProjectAuditAuthorizer) error { + if pluginID == "" || check == nil { + return extension.ErrProjectAuditUnsupported + } + s.mu.Lock() + defer s.mu.Unlock() + if s.check != nil || (s.holder != "" && s.holder != pluginID) { + return errors.New("extension: project audit authorizer already registered") + } + s.holder, s.check = pluginID, check + return nil +} + +func (s *projectAuditSlot) registerVerifier(pluginID string, verify extension.ProjectAuditLinkVerifier) (*auditPublisherToken, error) { + if pluginID == "" || verify == nil { + return nil, extension.ErrProjectAuditUnsupported + } + s.mu.Lock() + defer s.mu.Unlock() + if s.verifier != nil || (s.holder != "" && s.holder != pluginID) { + return nil, errors.New("extension: project audit verifier already registered") + } + s.holder, s.verifier = pluginID, verify + s.publisherToken = &auditPublisherToken{marker: 1} + return s.publisherToken, nil +} + +func (s *projectAuditSlot) registerDisclosure(pluginID string, disclose extension.ProjectAuditDisclosureAuthorizer) error { + if pluginID == "" || disclose == nil { + return extension.ErrProjectAuditUnsupported + } + s.mu.Lock() + defer s.mu.Unlock() + if s.disclose != nil || (s.holder != "" && s.holder != pluginID) { + return errors.New("extension: project audit disclosure authorizer already registered") + } + s.holder, s.disclose = pluginID, disclose + return nil +} + +func (r compiledRegistry) RegisterProjectAuditAuthorizer(extension.ProjectAuditAuthorizer) error { + return errors.New("extension: declarative plugins cannot register project audit authorizers") +} + +func (r pluginRegistry) RegisterProjectAuditAuthorizer(check extension.ProjectAuditAuthorizer) error { + if r.caps.projectAudit == nil { + return extension.ErrProjectAuditUnsupported + } + return r.caps.projectAudit.register(r.pluginID, check) +} + +func (r recordRegistry) RegisterProjectAuditAuthorizer(check extension.ProjectAuditAuthorizer) error { + reg, ok := r.Registry.(extension.ProjectAuditRegistry) + if !ok { + return extension.ErrProjectAuditUnsupported + } + return reg.RegisterProjectAuditAuthorizer(check) +} + +func (r compiledRegistry) RegisterProjectAuditLinkVerifier(extension.ProjectAuditLinkVerifier) (extension.ProjectAuditPublisherFactory, error) { + return nil, errors.New("extension: declarative plugins cannot register project audit link verifiers") +} +func (r compiledRegistry) RegisterProjectAuditDisclosureAuthorizer(extension.ProjectAuditDisclosureAuthorizer) error { + return errors.New("extension: declarative plugins cannot register project audit disclosure authorizers") +} +func (r pluginRegistry) RegisterProjectAuditLinkVerifier(verify extension.ProjectAuditLinkVerifier) (extension.ProjectAuditPublisherFactory, error) { + if r.caps.projectAudit == nil { + return nil, extension.ErrProjectAuditUnsupported + } + token, err := r.caps.projectAudit.registerVerifier(r.pluginID, verify) + if err != nil { + return nil, err + } + return projectAuditPublisherFactory{caps: r.caps, pluginID: r.pluginID, token: token}, nil +} +func (r pluginRegistry) RegisterProjectAuditDisclosureAuthorizer(disclose extension.ProjectAuditDisclosureAuthorizer) error { + if r.caps.projectAudit == nil { + return extension.ErrProjectAuditUnsupported + } + return r.caps.projectAudit.registerDisclosure(r.pluginID, disclose) +} +func (r recordRegistry) RegisterProjectAuditLinkVerifier(verify extension.ProjectAuditLinkVerifier) (extension.ProjectAuditPublisherFactory, error) { + reg, ok := r.Registry.(extension.ProjectAuditLinkRegistry) + if !ok { + return nil, extension.ErrProjectAuditUnsupported + } + return reg.RegisterProjectAuditLinkVerifier(verify) +} +func (r recordRegistry) RegisterProjectAuditDisclosureAuthorizer(disclose extension.ProjectAuditDisclosureAuthorizer) error { + reg, ok := r.Registry.(extension.ProjectAuditLinkRegistry) + if !ok { + return extension.ErrProjectAuditUnsupported + } + return reg.RegisterProjectAuditDisclosureAuthorizer(disclose) +} + +var _ extension.ProjectAuditHost = inprocHost{} +var _ extension.ProjectAuditPublisher = projectAuditPublisher{} +var _ extension.ProjectAuditPublisherFactory = projectAuditPublisherFactory{} + +func (h inprocHost) publishProjectAuditLink(ctx context.Context, req extension.ProjectAuditLinkRequest, capability *auditPublisherToken) (extension.ProjectAuditLinkReceipt, error) { + if err := ctx.Err(); err != nil { + return extension.ProjectAuditLinkReceipt{}, err + } + if h.caps.projectAudit == nil || h.caps.st == nil { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditUnsupported + } + slot := h.caps.projectAudit + slot.mu.RLock() + holder, verify, registeredToken := slot.holder, slot.verifier, slot.publisherToken + slot.mu.RUnlock() + if holder != h.pluginID || verify == nil || registeredToken == nil || capability != registeredToken { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditUnsupported + } + if h.owner == "" || (h.actor != "" && h.actor != h.owner) || req.HostAuditOffset < 0 || req.ProjectID == "" || req.SourceKind != "domain-commit" || req.SourceID == "" || req.OperationID == "" || req.ResourceKind == "" || req.ResourceID == "" || req.DisclosureClass != "metadata" || req.IdempotencyKey == "" { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditDenied + } + // An explicit offset is only a locator. A historical collision or a forged + // offset must not turn a private command with the same operation ID into a + // published project receipt. + resolved, err := h.caps.st.CommittedAuditOffset(h.owner, string(req.ProjectID), req.OperationID) + if errors.Is(err, store.ErrNotFound) { + if req.HostAuditOffset == 0 { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditPending + } + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditDenied + } + if errors.Is(err, store.ErrConflict) { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditLinkConflict + } + if err != nil { + return extension.ProjectAuditLinkReceipt{}, err + } + if req.HostAuditOffset != 0 && req.HostAuditOffset != resolved { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditLinkConflict + } + req.HostAuditOffset = resolved + row, projectID, err := h.caps.st.CommandAuditForLink(h.owner, req.HostAuditOffset) + if errors.Is(err, store.ErrNotFound) { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditDenied + } + if err != nil { + return extension.ProjectAuditLinkReceipt{}, err + } + if projectID != string(req.ProjectID) || row.Envelope.Actor != h.owner || row.Envelope.OperationID != req.OperationID || string(row.Envelope.State) != "committed" || row.Envelope.RunID != req.RunID { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditDenied + } + if req.RunID != "" { + bound, err := h.caps.st.RunBoundToAudit(h.owner, projectID, row.Envelope.ConversationID, req.RunID) + if err != nil { + return extension.ProjectAuditLinkReceipt{}, err + } + if !bound { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditDenied + } + } + previous, lookupErr := h.caps.st.ProjectAuditLinkByKey(h.owner, req.IdempotencyKey) + if lookupErr == nil { + if previous.HostAuditOffset != req.HostAuditOffset || previous.ProjectID != string(req.ProjectID) || previous.SourceKind != req.SourceKind || previous.SourceID != req.SourceID || previous.OperationID != req.OperationID || previous.ResourceKind != req.ResourceKind || previous.ResourceID != req.ResourceID || previous.RunID != req.RunID || previous.DisclosureClass != req.DisclosureClass { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditLinkConflict + } + return extension.ProjectAuditLinkReceipt{LinkID: previous.ID, HostAuditOffset: previous.HostAuditOffset}, nil + } + if !errors.Is(lookupErr, store.ErrNotFound) { + return extension.ProjectAuditLinkReceipt{}, lookupErr + } + proof := extension.ProjectAuditLinkProof{ProjectAuditLinkRequest: req, Owner: h.owner} + ok, err := verify(ctx, proof) + if err != nil { + return extension.ProjectAuditLinkReceipt{}, err + } + if !ok { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditDenied + } + if err := ctx.Err(); err != nil { + return extension.ProjectAuditLinkReceipt{}, err + } + link := store.ProjectAuditLink{HostAuditOffset: req.HostAuditOffset, ProjectID: projectID, Owner: h.owner, + SourceKind: req.SourceKind, SourceID: req.SourceID, OperationID: req.OperationID, ResourceKind: req.ResourceKind, + ResourceID: req.ResourceID, RunID: req.RunID, DisclosureClass: req.DisclosureClass, IdempotencyKey: req.IdempotencyKey} + saved, err := h.caps.st.SaveProjectAuditLink(link) + if errors.Is(err, store.ErrConflict) { + return extension.ProjectAuditLinkReceipt{}, extension.ErrProjectAuditLinkConflict + } + if err != nil { + return extension.ProjectAuditLinkReceipt{}, err + } + return extension.ProjectAuditLinkReceipt{LinkID: saved.ID, HostAuditOffset: saved.HostAuditOffset}, nil +} + +func (h inprocHost) QueryProjectAudit(ctx context.Context, q extension.ProjectAuditQuery) (extension.ProjectAuditPage, error) { + if err := ctx.Err(); err != nil { + return extension.ProjectAuditPage{}, err + } + if h.actor == "" || h.actor != h.owner || q.ProjectID == "" { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditDenied + } + if h.caps.projectAudit == nil || h.caps.st == nil { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditUnsupported + } + slot := h.caps.projectAudit + slot.mu.RLock() + holder, check := slot.holder, slot.check + slot.mu.RUnlock() + if holder != h.pluginID || check == nil { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditUnsupported + } + ok, err := check(h.actor, core.ID(q.ProjectID)) + if err != nil { + return extension.ProjectAuditPage{}, err + } + if !ok { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditDenied + } + if err := ctx.Err(); err != nil { + return extension.ProjectAuditPage{}, err + } + if q.LinkedSourceKind != "" || q.LinkedSourceID != "" || q.LinkedHostOffset != 0 { + return h.queryLinkedProjectAudit(ctx, q, slot) + } + rows, cutoff, next, err := h.caps.st.ListProjectCommandAudits(h.owner, string(q.ProjectID), q.BeforeOffset, q.CutoffOffset, q.Limit) + if err != nil { + return extension.ProjectAuditPage{}, err + } + page := extension.ProjectAuditPage{ + Receipts: make([]extension.HostAuditReceipt, 0, len(rows)), + NextOffset: next, CutoffOffset: cutoff, + Coverage: extension.ProjectAuditCoverage{ + CommandAudit: "actor-owned-project-bound-only", CutoffConsistency: "committed-prefix", CrossMember: "unsupported", LegacyProjectBinding: "not-recorded", RunRelations: "unsupported", ModelReceipts: "unsupported", + BackgroundReceipts: "unsupported", ResourceBodies: "unsupported", + }, + } + for _, row := range rows { + nameStatus := "not-recorded" + if row.ActorName != "" { + nameStatus = "recorded" + } + page.Receipts = append(page.Receipts, extension.HostAuditReceipt{ + SourceOffset: row.Offset, Time: row.Envelope.Time, ActorID: row.Envelope.Actor, + ActorName: row.ActorName, ActorNameStatus: nameStatus, + CommandID: row.Envelope.CommandID, State: string(row.Envelope.State), + OperationID: row.Envelope.OperationID, ConversationID: row.Envelope.ConversationID, + }) + } + return page, nil +} + +func (h inprocHost) queryLinkedProjectAudit(ctx context.Context, q extension.ProjectAuditQuery, slot *projectAuditSlot) (extension.ProjectAuditPage, error) { + if q.LinkedSourceKind != "domain-commit" || q.LinkedSourceID == "" || q.LinkedHostOffset <= 0 || q.BeforeOffset != 0 || q.CutoffOffset != 0 { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditUnsupported + } + slot.mu.RLock() + disclose := slot.disclose + slot.mu.RUnlock() + if disclose == nil { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditUnsupported + } + link, row, err := h.caps.st.ProjectAuditLinkedReceipt(string(q.ProjectID), q.LinkedSourceKind, q.LinkedSourceID, q.LinkedHostOffset) + if errors.Is(err, store.ErrNotFound) { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditDenied + } + if err != nil { + return extension.ProjectAuditPage{}, err + } + resource := extension.ProjectAuditLinkedResource{ProjectID: q.ProjectID, Owner: link.Owner, SourceKind: link.SourceKind, SourceID: link.SourceID, + OperationID: link.OperationID, ResourceKind: link.ResourceKind, ResourceID: link.ResourceID, RunID: link.RunID, DisclosureClass: link.DisclosureClass} + allowed, err := disclose(ctx, h.actor, resource) + if err != nil { + return extension.ProjectAuditPage{}, err + } + if !allowed.Metadata { + return extension.ProjectAuditPage{}, extension.ErrProjectAuditDenied + } + nameStatus := "not-recorded" + if row.ActorName != "" { + nameStatus = "recorded" + } + return extension.ProjectAuditPage{Receipts: []extension.HostAuditReceipt{{SourceOffset: row.Offset, Time: row.Envelope.Time, + ActorID: row.Envelope.Actor, ActorName: row.ActorName, ActorNameStatus: nameStatus, CommandID: row.Envelope.CommandID, + State: string(row.Envelope.State), OperationID: row.Envelope.OperationID}}, + Coverage: extension.ProjectAuditCoverage{CommandAudit: "verified-exact-link-only", CrossMember: "verified-exact-link-only", + CutoffConsistency: "exact-link", LegacyProjectBinding: "not-recorded", RunRelations: "unsupported", + ModelReceipts: "unsupported", BackgroundReceipts: "unsupported", ResourceBodies: "unsupported"}}, nil +} diff --git a/internal/app/project_audit_link_test.go b/internal/app/project_audit_link_test.go new file mode 100644 index 0000000..8668fed --- /dev/null +++ b/internal/app/project_audit_link_test.go @@ -0,0 +1,184 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + agentcommand "agent-command" + "agent-system-backend/internal/store" + "agent-system-backend/pkg/extension" + core "agent-system-core" +) + +func TestProjectAuditLinkSharesOnlyVerifiedCommittedMetadata(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "audit.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnsureUser("alice", "Alice at action"); err != nil { + t.Fatal(err) + } + if err := st.EnsureUser("bob", "Bob"); err != nil { + t.Fatal(err) + } + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + otherProject, err := st.CreateProject("alice", "B") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "private", "") + if err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "rdm.confirm", OperationID: "op-1", ConversationID: string(c.ID), State: agentcommand.StateCommitted, ParamsDigest: "secret-digest"}); err != nil { + t.Fatal(err) + } + private, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + offset := private[0].Offset + slot := &projectAuditSlot{} + caps := hostcaps{st: st, projectAudit: slot} + reg := pluginRegistry{pluginID: "rdm", caps: caps} + auditAllowed := true + if err := reg.RegisterProjectAuditAuthorizer(func(actor string, projectID core.ID) (bool, error) { + return auditAllowed && actor == "bob" && projectID == p.ID, nil + }); err != nil { + t.Fatal(err) + } + publisherFactory, err := (recordRegistry{Registry: reg}).RegisterProjectAuditLinkVerifier(func(_ context.Context, proof extension.ProjectAuditLinkProof) (bool, error) { + return proof.Owner == "alice" && proof.ProjectID == p.ID && proof.SourceKind == "domain-commit" && proof.SourceID == "commit-1" && proof.OperationID == "op-1" && proof.ResourceKind == "document" && proof.ResourceID == "doc-1", nil + }) + if err != nil { + t.Fatal(err) + } + metadataAllowed := true + if err := reg.RegisterProjectAuditDisclosureAuthorizer(func(_ context.Context, actor string, resource extension.ProjectAuditLinkedResource) (extension.ProjectAuditDisclosure, error) { + return extension.ProjectAuditDisclosure{Metadata: metadataAllowed && actor == "bob" && resource.ResourceID == "doc-1"}, nil + }); err != nil { + t.Fatal(err) + } + bob := inprocHost{caps: caps, pluginID: "rdm", owner: "bob", actor: "bob"} + q := extension.ProjectAuditQuery{ProjectID: p.ID, LinkedSourceKind: "domain-commit", LinkedSourceID: "commit-1", LinkedHostOffset: offset} + if _, err := bob.QueryProjectAudit(context.Background(), q); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("unlinked private audit: %v", err) + } + alicePublisher, err := publisherFactory.ForOwner("alice") // capability retained by trusted RDM outbox + if err != nil { + t.Fatal(err) + } + if _, ok := alicePublisher.(extension.Host); ok { + t.Fatal("publisher factory exposed a full owner-scoped Host") + } + linkReq := extension.ProjectAuditLinkRequest{ProjectID: p.ID, SourceKind: "domain-commit", SourceID: "commit-1", OperationID: "op-1", ResourceKind: "document", ResourceID: "doc-1", DisclosureClass: "metadata", IdempotencyKey: "outbox-1"} + // A different compiled plugin can call the public scoped-host factory with + // "rdm" as text. That must not mint RDM's audit publication authority. + wireScopedHostFactory(caps) + t.Cleanup(func() { extension.SetScopedHostFactory(nil) }) + forged, err := extension.NewScopedHost("alice", "rdm") + if err != nil { + t.Fatal(err) + } + if publisher, ok := forged.(extension.ProjectAuditPublisher); ok { + if _, err := publisher.PublishProjectAuditLink(context.Background(), linkReq); !errors.Is(err, extension.ErrProjectAuditUnsupported) { + t.Fatalf("self-reported plugin ID minted audit publisher: %v", err) + } + } + if rows, err := bob.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: p.ID}); err != nil || len(rows.Receipts) != 0 { + t.Fatalf("private owner row visible without exact link: %+v %v", rows, err) + } + wrongProject := linkReq + wrongProject.ProjectID = otherProject.ID + wrongProject.HostAuditOffset = offset + if _, err := alicePublisher.PublishProjectAuditLink(context.Background(), wrongProject); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("wrong host project: %v", err) + } + uncommitted := linkReq + uncommitted.SourceID, uncommitted.IdempotencyKey = "uncommitted", "outbox-uncommitted" + if _, err := alicePublisher.PublishProjectAuditLink(context.Background(), uncommitted); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("uncommitted source: %v", err) + } + if _, err := (pluginRegistry{pluginID: "other", caps: caps}).RegisterProjectAuditLinkVerifier(func(context.Context, extension.ProjectAuditLinkProof) (bool, error) { return true, nil }); err == nil { + t.Fatal("other plugin replaced the registered verifier") + } + pending := linkReq + pending.OperationID, pending.IdempotencyKey = "op-not-yet-recorded", "outbox-pending" + if _, err := alicePublisher.PublishProjectAuditLink(context.Background(), pending); !errors.Is(err, extension.ErrProjectAuditPending) { + t.Fatalf("host receipt race must be retryable: %v", err) + } + first, err := alicePublisher.PublishProjectAuditLink(context.Background(), linkReq) + if err != nil { + t.Fatal(err) + } + if first.HostAuditOffset != offset { + t.Fatalf("trusted resolved offset: got %d want %d", first.HostAuditOffset, offset) + } + retry, err := alicePublisher.PublishProjectAuditLink(context.Background(), linkReq) + if err != nil || retry.LinkID != first.LinkID { + t.Fatalf("idempotent retry: first=%+v retry=%+v err=%v", first, retry, err) + } + page, err := bob.QueryProjectAudit(context.Background(), q) + if err != nil { + t.Fatal(err) + } + if len(page.Receipts) != 1 || page.Receipts[0].CommandID != "rdm.confirm" || page.Receipts[0].ActorName != "Alice at action" || page.Receipts[0].ConversationID != "" || page.Coverage.CrossMember != "verified-exact-link-only" { + t.Fatalf("linked page: %+v", page) + } + raw, err := json.Marshal(page) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(raw), "secret-digest") || strings.Contains(string(raw), string(c.ID)) { + t.Fatalf("private receipt fields leaked: %s", raw) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "private.note", OperationID: "op-1", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + duplicates, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + privateOffset := duplicates[0].Offset + mislink := linkReq + mislink.HostAuditOffset, mislink.IdempotencyKey = privateOffset, "outbox-private" + if _, err := alicePublisher.PublishProjectAuditLink(context.Background(), mislink); !errors.Is(err, extension.ErrProjectAuditLinkConflict) { + t.Fatalf("ambiguous operation linked a private command: %v", err) + } + privateQuery := q + privateQuery.LinkedHostOffset = privateOffset + if _, err := bob.QueryProjectAudit(context.Background(), privateQuery); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("private command disclosed through ambiguous operation: %v", err) + } + metadataAllowed = false + if _, err := bob.QueryProjectAudit(context.Background(), q); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("revoked object metadata: %v", err) + } + metadataAllowed, auditAllowed = true, false + if _, err := bob.QueryProjectAudit(context.Background(), q); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("revoked audit.read: %v", err) + } + linkReq.ResourceID = "doc-2" + if _, err := alicePublisher.PublishProjectAuditLink(context.Background(), linkReq); !errors.Is(err, extension.ErrProjectAuditLinkConflict) { + t.Fatalf("same key different link: %v", err) + } + metadataAllowed, auditAllowed = true, true + if err := st.Close(); err != nil { + t.Fatal(err) + } + if _, err := alicePublisher.PublishProjectAuditLink(context.Background(), linkReq); err == nil || errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("storage outage during publish must remain retryable: %v", err) + } + if _, err := bob.QueryProjectAudit(context.Background(), q); err == nil || errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("storage outage during linked read must be explicit: %v", err) + } +} diff --git a/internal/app/project_audit_test.go b/internal/app/project_audit_test.go new file mode 100644 index 0000000..77a5e18 --- /dev/null +++ b/internal/app/project_audit_test.go @@ -0,0 +1,238 @@ +package app + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + agentcommand "agent-command" + "agent-system-backend/internal/store" + "agent-system-backend/pkg/extension" + core "agent-system-core" +) + +func TestProjectAuditHostChecksDomainReadAndProjectsStoredReceipts(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "audit.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + for _, owner := range []string{"alice", "bob"} { + if err := st.EnsureUser(owner, strings.ToUpper(owner)); err != nil { + t.Fatal(err) + } + } + a, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + b, err := st.CreateProject("alice", "B") + if err != nil { + t.Fatal(err) + } + ac, err := st.CreateConversation("alice", string(a.ID), "A", "") + if err != nil { + t.Fatal(err) + } + bc, err := st.CreateConversation("alice", string(b.ID), "B", "") + if err != nil { + t.Fatal(err) + } + shared, err := st.CreateConversationInProject("bob", string(a.ID), "A shared", "") + if err != nil { + t.Fatal(err) + } + for _, item := range []struct{ owner, conversation, command string }{ + {"alice", string(ac.ID), "a.command"}, + {"alice", string(bc.ID), "b.command"}, + {"bob", string(shared.ID), "shared.command"}, + } { + if err := st.AppendAudit(item.owner, agentcommand.AuditEnvelope{Time: time.Now(), Actor: item.owner, CommandID: item.command, ConversationID: item.conversation, State: agentcommand.StateCommitted, ParamsDigest: "sensitive-digest"}); err != nil { + t.Fatal(err) + } + } + slot := &projectAuditSlot{} + caps := hostcaps{st: st, projectAudit: slot} + reg := pluginRegistry{pluginID: "rdm", caps: caps} + allowed := true + if err := reg.RegisterProjectAuditAuthorizer(func(actor string, projectID core.ID) (bool, error) { + return allowed && actor == "bob" && projectID == a.ID, nil + }); err != nil { + t.Fatal(err) + } + h := inprocHost{caps: caps, owner: "bob", actor: "bob", pluginID: "rdm"} + page, err := h.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: a.ID, Limit: 10}) + if err != nil { + t.Fatal(err) + } + if len(page.Receipts) != 1 || page.Receipts[0].CommandID != "shared.command" { + t.Fatalf("A page: %+v", page) + } + if page.Coverage.CommandAudit != "actor-owned-project-bound-only" || page.Coverage.CrossMember != "unsupported" || page.Coverage.LegacyProjectBinding != "not-recorded" || page.Coverage.RunRelations != "unsupported" || page.Coverage.ModelReceipts != "unsupported" || page.Coverage.CutoffConsistency != "committed-prefix" { + t.Fatalf("coverage: %+v", page.Coverage) + } + if page.Receipts[0].ActorName != "BOB" { + t.Fatalf("actor names: %+v", page.Receipts) + } + raw, err := json.Marshal(page) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(raw), "sensitive-digest") || strings.Contains(string(raw), "paramsDigest") || strings.Contains(string(raw), "payload") { + t.Fatalf("raw audit payload leaked: %s", raw) + } + if _, err := h.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: b.ID}); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("project B: %v", err) + } + if _, err := (inprocHost{caps: caps, owner: "bob", actor: "bob", pluginID: "other"}).QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: a.ID}); !errors.Is(err, extension.ErrProjectAuditUnsupported) { + t.Fatalf("other plugin: %v", err) + } + if _, err := (inprocHost{caps: caps, owner: "bob", pluginID: "rdm"}).QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: a.ID}); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("unbound actor: %v", err) + } + allowed = false + if _, err := h.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: a.ID, BeforeOffset: page.NextOffset, CutoffOffset: page.CutoffOffset}); !errors.Is(err, extension.ErrProjectAuditDenied) { + t.Fatalf("revoked audit.read: %v", err) + } +} + +// A direct explicit ID bypasses the v39 default allocator. This must never +// yield a committed-prefix page that permanently skips the late lower row. +func TestProjectAuditPostgresExplicitLowIDCannotHideAcrossPages(t *testing.T) { + dsn := os.Getenv("AGENT_PG_TEST_DSN") + if dsn == "" { + t.Skip("AGENT_PG_TEST_DSN unset: isolated PostgreSQL check NOT_RUN") + } + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { admin.Close() }) + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil { + t.Fatal(err) + } + schema := "w9_" + strings.ToLower(string(store.NewID("audit"))) + if _, err := admin.Exec(`CREATE SCHEMA ` + schema); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { admin.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`) }) + st, err := store.OpenPostgres(store.AppendPGRuntimeParam(dsn, "search_path", schema+",public"), filepath.Join(t.TempDir(), "artifacts")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnsureUser("alice", "Alice"); err != nil { + t.Fatal(err) + } + project, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + conversation, err := st.CreateConversation("alice", string(project.ID), "work", "") + if err != nil { + t.Fatal(err) + } + for _, command := range []string{"first", "second"} { + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: command, ConversationID: string(conversation.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + } + insert := `INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload,actor_name,project_id) + VALUES ($1,$2,'','','committed','',$3,'alice',$4,'Alice',$5) RETURNING id` + gap, err := st.DB().Begin() + if err != nil { + t.Fatal(err) + } + defer gap.Rollback() + var gapID int64 + if err := gap.QueryRow(insert, time.Now().UTC().Format(time.RFC3339Nano), "rolled-back-gap", string(conversation.ID), `{}`, string(project.ID)).Scan(&gapID); err != nil { + t.Fatal(err) + } + if gapID != 3 { + t.Fatalf("rollback gap ID=%d want=3", gapID) + } + if err := gap.Rollback(); err != nil { + t.Fatal(err) + } + a, err := st.DB().Begin() + if err != nil { + t.Fatal(err) + } + defer a.Rollback() + env := agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "explicit-late", ConversationID: string(conversation.ID), State: agentcommand.StateCommitted} + raw, err := json.Marshal(env) + if err != nil { + t.Fatal(err) + } + _, err = a.Exec(`INSERT INTO command_audit(id,time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload,actor_name,project_id) + VALUES (3,$1,$2,'','','committed','',$3,'alice',$4,'Alice',$5)`, env.Time.UTC().Format(time.RFC3339Nano), env.CommandID, string(conversation.ID), string(raw), string(project.ID)) + if err != nil { + // Rejecting the online explicit ID is also a safe outcome. + t.Logf("online explicit low ID rejected: %v", err) + return + } + slot := &projectAuditSlot{} + caps := hostcaps{st: st, projectAudit: slot} + reg := pluginRegistry{pluginID: "rdm", caps: caps} + if err := reg.RegisterProjectAuditAuthorizer(func(actor string, projectID core.ID) (bool, error) { + return actor == "alice" && projectID == project.ID, nil + }); err != nil { + t.Fatal(err) + } + h := inprocHost{caps: caps, owner: "alice", actor: "alice", pluginID: "rdm"} + committed := make(chan error, 1) + started := make(chan struct{}) + go func() { + close(started) + committed <- st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "implicit-high", ConversationID: string(conversation.ID), State: agentcommand.StateCommitted}) + }() + <-started + var first extension.ProjectAuditPage + highCommittedEarly := false + select { + case err := <-committed: + if err != nil { + t.Fatal(err) + } + highCommittedEarly = true + case <-time.After(500 * time.Millisecond): + } + if highCommittedEarly { + first, err = h.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: project.ID, Limit: 2}) + if err != nil { + t.Fatal(err) + } + if len(first.Receipts) != 2 || first.Receipts[0].SourceOffset != 4 || first.Receipts[1].SourceOffset != 2 || first.CutoffOffset != 4 || first.NextOffset != 2 || first.Coverage.CutoffConsistency != "committed-prefix" { + t.Fatalf("counterexample first page changed: %+v", first) + } + } + if err := a.Commit(); err != nil { + t.Fatal(err) + } + if !highCommittedEarly { + if err := <-committed; err != nil { + t.Fatal(err) + } + first, err = h.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: project.ID, Limit: 2}) + if err != nil { + t.Fatal(err) + } + } + second, err := h.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: project.ID, BeforeOffset: first.NextOffset, CutoffOffset: first.CutoffOffset, Limit: 2}) + if err != nil { + t.Fatal(err) + } + seen := map[int64]bool{} + for _, receipt := range append(first.Receipts, second.Receipts...) { + seen[receipt.SourceOffset] = true + } + if !seen[3] || len(seen) != 4 { + t.Fatalf("committed-prefix omitted explicit low ID 3: early=%v first=%+v second=%+v", highCommittedEarly, first, second) + } +} diff --git a/internal/app/surface_audit_actor_test.go b/internal/app/surface_audit_actor_test.go new file mode 100644 index 0000000..b4f3fb3 --- /dev/null +++ b/internal/app/surface_audit_actor_test.go @@ -0,0 +1,115 @@ +package app + +import ( + "context" + "encoding/json" + "path/filepath" + "testing" + "time" + + agentcommand "agent-command" + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/store" + "agent-system-backend/internal/surfacesvc" + "agent-system-backend/pkg/extension" + core "agent-system-core" +) + +func TestSurfaceButtonDispatchUsesAuthenticatedActor(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "audit.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnsureUser("alice", "Alice at action"); err != nil { + t.Fatal(err) + } + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + bus := cmdbus.New(st, nil, func() time.Time { return time.Now().UTC() }) + if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.submit", Title: "Submit"}, "", func(context.Context, cmdbus.Request) (any, error) { return "ok", nil }); err != nil { + t.Fatal(err) + } + result, err := dispatchSurfaceAction(context.Background(), bus, surfacesvc.ActionDispatch{CommandID: "demo.submit", Owner: "alice", ConversationID: string(c.ID)}) + if err != nil { + t.Fatal(err) + } + if result.State != "committed" { + t.Fatalf("state: %s", result.State) + } + rows, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Envelope.Actor != "alice" || rows[0].ActorName != "Alice at action" { + t.Fatalf("button audit actor: %+v", rows) + } +} + +func TestCompiledHandlerReceivesHostOperationID(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "audit.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + bus := cmdbus.New(st, nil, func() time.Time { return time.Now().UTC() }) + slot := &projectAuditSlot{} + caps := hostcaps{st: st, projectAudit: slot} + reg := pluginRegistry{compiledRegistry: compiledRegistry{bus: bus}, pluginID: "rdm", caps: caps} + var seen string + if err := reg.AddCommandHandler(json.RawMessage(`{"id":"rdm.confirm","title":"Confirm"}`), func(_ context.Context, _ extension.Host, req extension.Request) (any, error) { + seen = req.OperationID + return "domain-committed", nil + }); err != nil { + t.Fatal(err) + } + resp := bus.Dispatch(context.Background(), cmdbus.Request{ID: "rdm.confirm", Actor: "alice", Owner: "alice", ConversationID: string(c.ID)}) + rows, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + if seen == "" || resp.Receipt.OperationID != seen || len(rows) != 1 || rows[0].Envelope.OperationID != seen { + t.Fatalf("handler/domain correlation: seen=%q receipt=%+v rows=%+v", seen, resp.Receipt, rows) + } + if err := reg.RegisterProjectAuditAuthorizer(func(actor string, project core.ID) (bool, error) { return actor == "bob" && project == p.ID, nil }); err != nil { + t.Fatal(err) + } + publisherFactory, err := (recordRegistry{Registry: reg}).RegisterProjectAuditLinkVerifier(func(_ context.Context, proof extension.ProjectAuditLinkProof) (bool, error) { + return proof.Owner == "alice" && proof.ProjectID == p.ID && proof.OperationID == seen && proof.SourceID == "domain-commit-1" && proof.ResourceID == "doc-1", nil + }) + if err != nil { + t.Fatal(err) + } + if err := reg.RegisterProjectAuditDisclosureAuthorizer(func(_ context.Context, actor string, r extension.ProjectAuditLinkedResource) (extension.ProjectAuditDisclosure, error) { + return extension.ProjectAuditDisclosure{Metadata: actor == "bob" && r.ResourceID == "doc-1"}, nil + }); err != nil { + t.Fatal(err) + } + publisher, err := publisherFactory.ForOwner("alice") + if err != nil { + t.Fatal(err) + } + linked, err := publisher.PublishProjectAuditLink(context.Background(), extension.ProjectAuditLinkRequest{ProjectID: p.ID, OperationID: seen, SourceKind: "domain-commit", SourceID: "domain-commit-1", ResourceKind: "document", ResourceID: "doc-1", DisclosureClass: "metadata", IdempotencyKey: "outbox-1"}) + if err != nil || linked.HostAuditOffset != rows[0].Offset { + t.Fatalf("publish by trusted operation: %+v %v", linked, err) + } + bob := inprocHost{caps: caps, owner: "bob", actor: "bob", pluginID: "rdm"} + page, err := bob.QueryProjectAudit(context.Background(), extension.ProjectAuditQuery{ProjectID: p.ID, LinkedSourceKind: "domain-commit", LinkedSourceID: "domain-commit-1", LinkedHostOffset: linked.HostAuditOffset}) + if err != nil || len(page.Receipts) != 1 || page.Receipts[0].OperationID != seen { + t.Fatalf("linked read: %+v %v", page, err) + } +} diff --git a/internal/app/surface_dispatch.go b/internal/app/surface_dispatch.go new file mode 100644 index 0000000..7a8137a --- /dev/null +++ b/internal/app/surface_dispatch.go @@ -0,0 +1,25 @@ +package app + +import ( + "context" + + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/surfacesvc" +) + +// dispatchSurfaceAction receives Owner from SurfaceService's authenticated +// caller. The button is acting for that user, so both Actor and Owner must be +// populated before the command bus writes its audit envelope. +func dispatchSurfaceAction(ctx context.Context, bus *cmdbus.Bus, d surfacesvc.ActionDispatch) (surfacesvc.ActionOutcome, error) { + resp := bus.Dispatch(ctx, cmdbus.Request{ID: d.CommandID, Params: d.Params, + Entrance: "button", Actor: d.Owner, Owner: d.Owner, ConversationID: d.ConversationID}) + state := "committed" + if resp.Receipt.State != "" { + state = string(resp.Receipt.State) + } + cardID := "" + if resp.Card != nil { + cardID = string(resp.Card.ID) + } + return surfacesvc.ActionOutcome{State: state, CardID: cardID, Receipt: resp.Receipt, Payload: resp.Payload}, nil +} diff --git a/internal/cmdbus/cmdbus.go b/internal/cmdbus/cmdbus.go index 9b63e7d..3fb4a14 100644 --- a/internal/cmdbus/cmdbus.go +++ b/internal/cmdbus/cmdbus.go @@ -5,6 +5,7 @@ package cmdbus import ( "context" + "crypto/rand" "crypto/sha256" "encoding/hex" "encoding/json" @@ -89,6 +90,7 @@ type Request struct { Actor string Owner string ConversationID string + OperationID string // overwritten by the trusted bus before handler execution AnchorMessageID string // 确认卡锚定的发起消息（内联渲染） Confirmed bool // 危险命令经确认卡放行后的执行 HostProjectID string // EXT-18：分发所属基座项目（空＝无项目上下文） @@ -279,12 +281,19 @@ func (b *Bus) Dispatch(ctx context.Context, req Request) Response { if !ok { return receipt(agentcommand.StateRejected, "", "no host handler bound for "+req.ID) } + // The domain handler can persist this trusted ID with its transaction/outbox. + // Entropy, not the clock or caller Params, distinguishes concurrent operations. + // A handler failure still writes only a rejected audit receipt without an ID. + var opBytes [24]byte + if _, err := rand.Read(opBytes[:]); err != nil { + return receipt(agentcommand.StateUnknown, "", "operation ID unavailable") + } + req.OperationID = "op_" + hex.EncodeToString(opBytes[:]) payload, err := h(ctx, req) if err != nil { return receipt(agentcommand.StateRejected, "", "rejected: "+err.Error()) } - op := fmt.Sprintf("op_%s_%d", strings.ReplaceAll(strings.TrimPrefix(req.ID, ""), ".", "_"), b.now().UnixNano()) - return Response{Receipt: b.auditReceipt(req, agentcommand.StateCommitted, op, "", start), Payload: payload} + return Response{Receipt: b.auditReceipt(req, agentcommand.StateCommitted, req.OperationID, "", start), Payload: payload} } // Confirm 确认卡回道：actionID=confirm 执行挂起请求（entrance 归一 card），其余视为取消。 diff --git a/internal/cmdbus/cmdbus_test.go b/internal/cmdbus/cmdbus_test.go index 7b4e3f3..90d5c63 100644 --- a/internal/cmdbus/cmdbus_test.go +++ b/internal/cmdbus/cmdbus_test.go @@ -87,6 +87,66 @@ func TestDispatchFourEntrancesSameAudit(t *testing.T) { } } +func TestHandlerReceivesCommittedOperationIDBeforeDomainWrite(t *testing.T) { + audit := &memAudit{} + bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 123).UTC() }) + var seen string + if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "rdm.confirm", Title: "Confirm"}, "", func(_ context.Context, req Request) (any, error) { + seen = req.OperationID + return "committed-domain-event", nil + }); err != nil { + t.Fatal(err) + } + resp := bus.Dispatch(context.Background(), Request{ID: "rdm.confirm", Actor: "alice", Owner: "alice", Params: map[string]any{"operationId": "forged-client-id"}}) + if seen == "" || seen == "forged-client-id" || resp.Receipt.OperationID != seen || len(audit.logs) != 1 || audit.logs[0].OperationID != seen { + t.Fatalf("operation correlation missing: handler=%q receipt=%+v audit=%+v", seen, resp.Receipt, audit.logs) + } + if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "rdm.fail", Title: "Fail"}, "", func(_ context.Context, req Request) (any, error) { + if req.OperationID == "" { + t.Fatal("handler must receive a tentative host operation ID") + } + return nil, errors.New("domain transaction rejected") + }); err != nil { + t.Fatal(err) + } + failed := bus.Dispatch(context.Background(), Request{ID: "rdm.fail", Actor: "alice", Owner: "alice"}) + if failed.Receipt.State != agentcommand.StateRejected || failed.Receipt.OperationID != "" || len(audit.logs) != 2 || audit.logs[1].OperationID != "" { + t.Fatalf("failed handler was marked committed: %+v audits=%+v", failed.Receipt, audit.logs) + } +} + +func TestOperationIDsRemainUniqueWithFixedClockAndConcurrentDispatch(t *testing.T) { + audit := &memAudit{} + bus := New(audit, nil, func() time.Time { return time.Unix(1700000000, 123).UTC() }) + if err := bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "rdm.confirm", Title: "Confirm"}, "", func(_ context.Context, req Request) (any, error) { + return req.OperationID, nil + }); err != nil { + t.Fatal(err) + } + const count = 64 + var wg sync.WaitGroup + ids := make([]string, count) + for i := range ids { + wg.Add(1) + go func(i int) { + defer wg.Done() + resp := bus.Dispatch(context.Background(), Request{ID: "rdm.confirm", Actor: "alice", Owner: "alice"}) + ids[i] = resp.Receipt.OperationID + }(i) + } + wg.Wait() + seen := map[string]bool{} + for _, id := range ids { + if id == "" || seen[id] { + t.Fatalf("fixed clock produced a missing or duplicate trusted operation ID: %q", id) + } + seen[id] = true + } + if len(audit.logs) != count { + t.Fatalf("audit rows=%d want=%d", len(audit.logs), count) + } +} + func TestDispatchUnknownAndBadParams(t *testing.T) { a := &memAudit{} b := newTestBus(a, &memCards{}) diff --git a/internal/store/audit.go b/internal/store/audit.go index b5479a8..9b41a1c 100644 --- a/internal/store/audit.go +++ b/internal/store/audit.go @@ -3,6 +3,8 @@ package store import ( "database/sql" "encoding/json" + "errors" + "fmt" "time" agentcommand "agent-command" @@ -12,17 +14,98 @@ import ( // owner 是租户身份（认证注入），与信封里的 Actor（发起主体）语义不同：查询按 // owner 过滤，跨租户审计不可见。 func (s *Store) AppendAudit(owner string, env agentcommand.AuditEnvelope) error { + if err := s.requireProjectAuditCommittedPrefix(); err != nil { + return err + } raw, err := json.Marshal(env) if err != nil { return err } - _, err = s.db.Exec(s.q(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, owner, payload) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`), + _, err = s.db.Exec(s.q(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, owner, payload, actor_name, project_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE((SELECT name FROM users WHERE owner=?), ''), + COALESCE((SELECT project_id FROM conversations WHERE id=? AND owner=?), ''))`), env.Time.UTC().Format(time.RFC3339Nano), env.CommandID, string(env.Source), env.Entrance, string(env.State), - env.OperationID, env.ConversationID, owner, string(raw)) + env.OperationID, env.ConversationID, owner, string(raw), env.Actor, env.ConversationID, owner) return err } +// ProjectCommandAuditRow is an internal command receipt. Callers must project +// only approved fields; the envelope contains a digest and is not a public DTO. +type ProjectCommandAuditRow struct { + Offset int64 + ActorName string // empty on old rows or when no trusted account name was recorded + Envelope agentcommand.AuditEnvelope +} + +// ErrAuditCutoffUnsafe prevents a PostgreSQL page from claiming a fixed cutoff +// when the ID generator, sequence, or all-writer database guard cannot prove +// commit order. +var ErrAuditCutoffUnsafe = errors.New("store: project audit cutoff unsafe") + +func (s *Store) requireProjectAuditCommittedPrefix() error { + if s.dialect.name() != "postgres" { + return nil + } // SQLite serializes writers. + var safe bool + if err := s.db.QueryRow(`SELECT project_audit_cutoff_safe('command_audit'::regclass) AND project_audit_writers_guarded('command_audit'::regclass)`).Scan(&safe); err != nil { + return fmt.Errorf("%w: %v", ErrAuditCutoffUnsafe, err) + } + if !safe { + return ErrAuditCutoffUnsafe + } + return nil +} + +// ListProjectCommandAudits reads the project resolved from a trusted conversation +// at append time. Reassigning or deleting that conversation cannot rewrite the +// original project. Legacy rows without this snapshot are excluded: current +// project membership cannot establish their historical project attribution. +// Only the authenticated owner can read this first private slice. +func (s *Store) ListProjectCommandAudits(actorOwner, projectID string, before, cutoff int64, limit int) ([]ProjectCommandAuditRow, int64, int64, error) { + if actorOwner == "" || projectID == "" || before < 0 || cutoff < 0 || (before != 0 && cutoff == 0) { + return nil, 0, 0, errors.New("project audit: invalid scope or cursor") + } + if err := s.requireProjectAuditCommittedPrefix(); err != nil { + return nil, 0, 0, err + } + if limit <= 0 || limit > 200 { + limit = 50 + } + if cutoff == 0 { + if err := s.db.QueryRow(s.q(`SELECT COALESCE(MAX(id),0) FROM command_audit + WHERE project_id=? AND owner=?`), projectID, actorOwner).Scan(&cutoff); err != nil { + return nil, 0, 0, err + } + } + rows, err := s.db.Query(s.q(`SELECT id, actor_name, payload FROM command_audit + WHERE project_id=? AND owner=? AND id<=? AND (?=0 OR id1 OR cycles THEN RETURN false; END IF; + -- Supported writers allocate under the lock before they commit. Read + -- committed rows first, then the non-MVCC sequence state: a writer + -- committing between these reads can only advance the latter. + EXECUTE format('SELECT COALESCE(MIN(id),1),COALESCE(MAX(id),0) FROM %s',target) INTO min_id,max_id; + EXECUTE format('SELECT last_value,is_called FROM %s',seq) INTO last_number,was_called; + next_number := last_number; + IF was_called THEN next_number := last_number+increment_by; END IF; + RETURN min_id>0 AND next_number>0 AND next_number>max_id; + END $$`, + `CREATE FUNCTION next_command_audit_id(target regclass) RETURNS bigint LANGUAGE plpgsql VOLATILE AS $$ + DECLARE seq regclass; allocated bigint; max_id bigint; + BEGIN + PERFORM pg_advisory_xact_lock(715328491,target::oid::int4); + IF NOT project_audit_cutoff_safe(target) THEN RAISE EXCEPTION 'project audit cutoff sequence is unsafe'; END IF; + seq := pg_get_serial_sequence(target::text,'id')::regclass; + allocated := nextval(seq); + EXECUTE format('SELECT COALESCE(MAX(id),0) FROM %s',target) INTO max_id; + IF allocated<=0 OR allocated<=max_id THEN RAISE EXCEPTION 'project audit ID did not advance beyond committed rows'; END IF; + RETURN allocated; + END $$`, + `ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT next_command_audit_id('command_audit'::regclass)`, + `DO $$ BEGIN IF NOT project_audit_cutoff_safe('command_audit'::regclass) THEN RAISE EXCEPTION 'project audit cutoff migration precondition failed'; END IF; END $$`, + }, + }, + down: map[string][]string{ + "sqlite": {`SELECT 1`}, + "postgres": { + `LOCK TABLE command_audit IN ACCESS EXCLUSIVE MODE`, + `ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT nextval(pg_get_serial_sequence('command_audit','id')::regclass)`, + `DROP FUNCTION next_command_audit_id(regclass)`, + `DROP FUNCTION project_audit_cutoff_safe(regclass)`, + }, + }}, + migration{version: 40, name: "project_audit_all_writers_guard", + up: map[string][]string{ + "sqlite": {`SELECT 1`}, + "postgres": { + `LOCK TABLE command_audit IN ACCESS EXCLUSIVE MODE`, + `CREATE FUNCTION project_audit_insert_guard_v1() RETURNS trigger LANGUAGE plpgsql VOLATILE AS $$ + DECLARE seq regclass; last_number numeric; was_called boolean; max_id bigint; + BEGIN + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'command audit insert requires read committed isolation'; + END IF; + PERFORM pg_advisory_xact_lock(715328491,TG_RELID::int4); + IF NOT project_audit_cutoff_safe(TG_RELID) OR NOT project_audit_writers_guarded(TG_RELID) THEN + RAISE EXCEPTION 'command audit write guard is unsafe'; + END IF; + seq := pg_get_serial_sequence(TG_RELID::regclass::text,'id')::regclass; + EXECUTE format('SELECT last_value,is_called FROM %s',seq) INTO last_number,was_called; + EXECUTE format('SELECT COALESCE(MAX(id),0) FROM %s',TG_RELID::regclass) INTO max_id; + IF NEW.id IS NULL OR NEW.id<=0 OR NEW.id<=max_id OR NOT was_called OR NEW.id>last_number THEN + RAISE EXCEPTION 'command audit ID is not commit-monotone'; + END IF; + RETURN NEW; + END $$`, + `CREATE FUNCTION project_audit_mutation_guard_v1() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'command audit is append-only'; END $$`, + `CREATE TRIGGER project_audit_insert_guard_v1 BEFORE INSERT ON command_audit FOR EACH ROW EXECUTE FUNCTION project_audit_insert_guard_v1()`, + `CREATE TRIGGER project_audit_mutation_guard_v1 BEFORE UPDATE OR DELETE ON command_audit FOR EACH ROW EXECUTE FUNCTION project_audit_mutation_guard_v1()`, + `ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_insert_guard_v1`, + `ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_mutation_guard_v1`, + `CREATE FUNCTION project_audit_writers_guarded(target regclass) RETURNS boolean LANGUAGE plpgsql VOLATILE AS $$ + DECLARE insert_ok boolean; mutation_ok boolean; + BEGIN + SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid=target AND tgname='project_audit_insert_guard_v1' + AND NOT tgisinternal AND tgenabled='A' AND tgtype=7 + AND tgfoid=to_regprocedure('project_audit_insert_guard_v1()')) INTO insert_ok; + SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid=target AND tgname='project_audit_mutation_guard_v1' + AND NOT tgisinternal AND tgenabled='A' AND tgtype=27 + AND tgfoid=to_regprocedure('project_audit_mutation_guard_v1()')) INTO mutation_ok; + RETURN insert_ok AND mutation_ok; + END $$`, + `CREATE OR REPLACE FUNCTION next_command_audit_id(target regclass) RETURNS bigint LANGUAGE plpgsql VOLATILE AS $$ + DECLARE seq regclass; allocated bigint; max_id bigint; + BEGIN + PERFORM pg_advisory_xact_lock(715328491,target::oid::int4); + IF NOT project_audit_cutoff_safe(target) OR NOT project_audit_writers_guarded(target) THEN + RAISE EXCEPTION 'project audit cutoff guard is unsafe'; + END IF; + seq := pg_get_serial_sequence(target::text,'id')::regclass; + allocated := nextval(seq); + EXECUTE format('SELECT COALESCE(MAX(id),0) FROM %s',target) INTO max_id; + IF allocated<=0 OR allocated<=max_id THEN RAISE EXCEPTION 'project audit ID did not advance beyond committed rows'; END IF; + RETURN allocated; + END $$`, + `DO $$ BEGIN IF NOT project_audit_cutoff_safe('command_audit'::regclass) OR NOT project_audit_writers_guarded('command_audit'::regclass) THEN RAISE EXCEPTION 'project audit writer guard migration precondition failed'; END IF; END $$`, + }, + }, + down: map[string][]string{ + "sqlite": {`SELECT 1`}, + "postgres": { + `LOCK TABLE command_audit IN ACCESS EXCLUSIVE MODE`, + `CREATE OR REPLACE FUNCTION next_command_audit_id(target regclass) RETURNS bigint LANGUAGE plpgsql VOLATILE AS $$ + DECLARE seq regclass; allocated bigint; max_id bigint; + BEGIN + PERFORM pg_advisory_xact_lock(715328491,target::oid::int4); + IF NOT project_audit_cutoff_safe(target) THEN RAISE EXCEPTION 'project audit cutoff sequence is unsafe'; END IF; + seq := pg_get_serial_sequence(target::text,'id')::regclass; + allocated := nextval(seq); + EXECUTE format('SELECT COALESCE(MAX(id),0) FROM %s',target) INTO max_id; + IF allocated<=0 OR allocated<=max_id THEN RAISE EXCEPTION 'project audit ID did not advance beyond committed rows'; END IF; + RETURN allocated; + END $$`, + `DROP TRIGGER project_audit_insert_guard_v1 ON command_audit`, + `DROP TRIGGER project_audit_mutation_guard_v1 ON command_audit`, + `DROP FUNCTION project_audit_insert_guard_v1()`, + `DROP FUNCTION project_audit_mutation_guard_v1()`, + `DROP FUNCTION project_audit_writers_guarded(regclass)`, + }, + }}, ) } diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index 5f06402..774bc90 100644 --- a/internal/store/migrate_compat.go +++ b/internal/store/migrate_compat.go @@ -41,6 +41,10 @@ var migrationBreaksOldReaders = map[int]bool{ 34: false, // workflow_step_sources：新增逐步骤来源绑定表，旧读者不受影响 35: false, // context_manifest_resolve_scope：可信回执增加范围标记，旧读者不受影响 36: false, // scrub_candidate_context_scope_text：候选原文列清理，旧读者不受影响 + 37: false, // command_audit_attribution_snapshot：仅加列/索引，旧读者不受影响 + 38: false, // project_audit_committed_links：新表，旧读者不受影响 + 39: false, // project_audit_committed_prefix：PG 审计 ID 默认函数，旧读者兼容 + 40: false, // project_audit_all_writers_guard：PG 在线审计写入门禁，旧读者兼容 } // SchemaTarget 是本二进制会迁移到的 schema 版本。 diff --git a/internal/store/project_audit_links.go b/internal/store/project_audit_links.go new file mode 100644 index 0000000..6d5c96f --- /dev/null +++ b/internal/store/project_audit_links.go @@ -0,0 +1,175 @@ +package store + +import ( + "database/sql" + "encoding/json" + "errors" +) + +// ProjectAuditLink is an immutable association written only after a trusted +// domain verifier has checked the committed source event. +type ProjectAuditLink struct { + ID int64 + HostAuditOffset int64 + ProjectID string + Owner string + SourceKind string + SourceID string + OperationID string + ResourceKind string + ResourceID string + RunID string + DisclosureClass string + IdempotencyKey string +} + +// CommittedAuditOffset resolves exactly one persisted host receipt. A domain +// outbox can have the operation ID before the bus appends its audit row; no row +// yet is a retryable pending state, while duplicates are ambiguous and refused. +func (s *Store) CommittedAuditOffset(owner, projectID, operationID string) (int64, error) { + if owner == "" || projectID == "" || operationID == "" { + return 0, ErrNotFound + } + rows, err := s.db.Query(s.q(`SELECT id FROM command_audit WHERE owner=? AND project_id=? AND operation_id=? AND state='committed' ORDER BY id LIMIT 2`), owner, projectID, operationID) + if err != nil { + return 0, err + } + defer rows.Close() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return 0, err + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + return 0, err + } + if len(ids) == 0 { + return 0, ErrNotFound + } + if len(ids) > 1 { + return 0, ErrConflict + } + return ids[0], nil +} + +func (s *Store) CommandAuditForLink(owner string, offset int64) (ProjectCommandAuditRow, string, error) { + var row ProjectCommandAuditRow + var projectID, raw string + if owner == "" || offset <= 0 { + return row, "", ErrNotFound + } + err := s.db.QueryRow(s.q(`SELECT id,actor_name,project_id,payload FROM command_audit WHERE id=? AND owner=? AND project_id<>''`), offset, owner). + Scan(&row.Offset, &row.ActorName, &projectID, &raw) + if errors.Is(err, sql.ErrNoRows) { + return row, "", ErrNotFound + } + if err != nil { + return row, "", err + } + if err := json.Unmarshal([]byte(raw), &row.Envelope); err != nil { + return row, "", err + } + return row, projectID, nil +} + +func (s *Store) RunBoundToAudit(owner, projectID, conversationID, runID string) (bool, error) { + if owner == "" || projectID == "" || conversationID == "" || runID == "" { + return false, nil + } + var one int + err := s.db.QueryRow(s.q(`SELECT 1 FROM runs WHERE id=? AND owner=? AND project_id=? AND conversation_id=?`), runID, owner, projectID, conversationID).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + return err == nil, err +} + +const projectAuditLinkColumns = `id,host_audit_offset,project_id,owner,source_kind,source_id,operation_id,resource_kind,resource_id,run_id,disclosure_class,idempotency_key` +const projectAuditLinkAliasedColumns = `l.id,l.host_audit_offset,l.project_id,l.owner,l.source_kind,l.source_id,l.operation_id,l.resource_kind,l.resource_id,l.run_id,l.disclosure_class,l.idempotency_key` + +func scanProjectAuditLink(scan func(...any) error) (ProjectAuditLink, error) { + var link ProjectAuditLink + err := scan(&link.ID, &link.HostAuditOffset, &link.ProjectID, &link.Owner, &link.SourceKind, &link.SourceID, + &link.OperationID, &link.ResourceKind, &link.ResourceID, &link.RunID, &link.DisclosureClass, &link.IdempotencyKey) + if errors.Is(err, sql.ErrNoRows) { + return link, ErrNotFound + } + return link, err +} + +func (s *Store) ProjectAuditLinkByKey(owner, key string) (ProjectAuditLink, error) { + if owner == "" || key == "" { + return ProjectAuditLink{}, ErrNotFound + } + return scanProjectAuditLink(func(dst ...any) error { + return s.db.QueryRow(s.q(`SELECT `+projectAuditLinkColumns+` FROM project_audit_links WHERE owner=? AND idempotency_key=?`), owner, key).Scan(dst...) + }) +} + +func sameProjectAuditLink(a, b ProjectAuditLink) bool { + return a.HostAuditOffset == b.HostAuditOffset && a.ProjectID == b.ProjectID && a.Owner == b.Owner && + a.SourceKind == b.SourceKind && a.SourceID == b.SourceID && a.OperationID == b.OperationID && + a.ResourceKind == b.ResourceKind && a.ResourceID == b.ResourceID && a.RunID == b.RunID && + a.DisclosureClass == b.DisclosureClass && a.IdempotencyKey == b.IdempotencyKey +} + +// SaveProjectAuditLink is idempotent for an identical key and binding. Both the +// key and the host/source triple are unique; racing or conflicting retries +// resolve by reading the committed winner, never by overwriting it. +func (s *Store) SaveProjectAuditLink(link ProjectAuditLink) (ProjectAuditLink, error) { + var id int64 + err := s.db.QueryRow(s.q(`INSERT INTO project_audit_links(host_audit_offset,project_id,owner,source_kind,source_id,operation_id,resource_kind,resource_id,run_id,disclosure_class,idempotency_key,created_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT DO NOTHING RETURNING id`), link.HostAuditOffset, link.ProjectID, link.Owner, + link.SourceKind, link.SourceID, link.OperationID, link.ResourceKind, link.ResourceID, link.RunID, link.DisclosureClass, + link.IdempotencyKey, nowISO()).Scan(&id) + if err == nil { + link.ID = id + return link, nil + } + if !errors.Is(err, sql.ErrNoRows) { + return ProjectAuditLink{}, err + } + previous, err := scanProjectAuditLink(func(dst ...any) error { + return s.db.QueryRow(s.q(`SELECT `+projectAuditLinkColumns+` FROM project_audit_links + WHERE (owner=? AND idempotency_key=?) OR (host_audit_offset=? AND source_kind=? AND source_id=?) LIMIT 1`), + link.Owner, link.IdempotencyKey, link.HostAuditOffset, link.SourceKind, link.SourceID).Scan(dst...) + }) + if err != nil { + return ProjectAuditLink{}, err + } + if !sameProjectAuditLink(previous, link) { + return ProjectAuditLink{}, ErrConflict + } + return previous, nil +} + +// ProjectAuditLinkedReceipt reads one exact association and the underlying +// host row. It does not enumerate private links or decide field disclosure. +func (s *Store) ProjectAuditLinkedReceipt(projectID, sourceKind, sourceID string, hostOffset int64) (ProjectAuditLink, ProjectCommandAuditRow, error) { + var row ProjectCommandAuditRow + var link ProjectAuditLink + var raw string + if projectID == "" || sourceKind == "" || sourceID == "" || hostOffset <= 0 { + return link, row, ErrNotFound + } + err := s.db.QueryRow(s.q(`SELECT `+projectAuditLinkAliasedColumns+`,a.actor_name,a.payload FROM project_audit_links l + JOIN command_audit a ON a.id=l.host_audit_offset AND a.owner=l.owner AND a.project_id=l.project_id + WHERE l.project_id=? AND l.source_kind=? AND l.source_id=? AND l.host_audit_offset=?`), + projectID, sourceKind, sourceID, hostOffset).Scan(&link.ID, &link.HostAuditOffset, &link.ProjectID, &link.Owner, + &link.SourceKind, &link.SourceID, &link.OperationID, &link.ResourceKind, &link.ResourceID, &link.RunID, + &link.DisclosureClass, &link.IdempotencyKey, &row.ActorName, &raw) + if errors.Is(err, sql.ErrNoRows) { + return link, row, ErrNotFound + } + if err != nil { + return link, row, err + } + row.Offset = link.HostAuditOffset + if err := json.Unmarshal([]byte(raw), &row.Envelope); err != nil { + return link, row, err + } + return link, row, nil +} diff --git a/internal/store/project_audit_links_test.go b/internal/store/project_audit_links_test.go new file mode 100644 index 0000000..64b5b07 --- /dev/null +++ b/internal/store/project_audit_links_test.go @@ -0,0 +1,97 @@ +package store + +import ( + "errors" + "testing" + "time" + + agentcommand "agent-command" +) + +func TestProjectAuditLinkStoreIdempotenceAndExactScope(t *testing.T) { + st := openTestStore(t) // SQLite locally; isolated PostgreSQL schema when configured. + if err := st.EnsureUser("alice", "Alice"); err != nil { + t.Fatal(err) + } + a, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + b, err := st.CreateProject("alice", "B") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(a.ID), "work", "") + if err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "rdm.confirm", OperationID: "op-1", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + rows, _, _, err := st.ListProjectCommandAudits("alice", string(a.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + resolved, err := st.CommittedAuditOffset("alice", string(a.ID), "op-1") + if err != nil || resolved != rows[0].Offset { + t.Fatalf("trusted operation lookup: %d %v", resolved, err) + } + if _, err := st.CommittedAuditOffset("alice", string(a.ID), "op-pending"); !errors.Is(err, ErrNotFound) { + t.Fatalf("unrecorded operation should wait: %v", err) + } + link := ProjectAuditLink{HostAuditOffset: rows[0].Offset, ProjectID: string(a.ID), Owner: "alice", SourceKind: "domain-commit", SourceID: "commit-1", OperationID: "op-1", ResourceKind: "document", ResourceID: "doc-1", DisclosureClass: "metadata", IdempotencyKey: "outbox-1"} + first, err := st.SaveProjectAuditLink(link) + if err != nil { + t.Fatal(err) + } + retry, err := st.SaveProjectAuditLink(link) + if err != nil || retry.ID != first.ID { + t.Fatalf("retry: first=%+v retry=%+v err=%v", first, retry, err) + } + if found, err := st.ProjectAuditLinkByKey("alice", "outbox-1"); err != nil || found.ID != first.ID { + t.Fatalf("key lookup: %+v %v", found, err) + } + if _, _, err := st.ProjectAuditLinkedReceipt(string(b.ID), "domain-commit", "commit-1", rows[0].Offset); !errors.Is(err, ErrNotFound) { + t.Fatalf("wrong project: %v", err) + } + if _, _, err := st.ProjectAuditLinkedReceipt(string(a.ID), "domain-commit", "other", rows[0].Offset); !errors.Is(err, ErrNotFound) { + t.Fatalf("wrong source: %v", err) + } + got, audit, err := st.ProjectAuditLinkedReceipt(string(a.ID), "domain-commit", "commit-1", rows[0].Offset) + if err != nil || got.ID != first.ID || audit.Envelope.CommandID != "rdm.confirm" { + t.Fatalf("linked read: %+v %+v %v", got, audit, err) + } + link.ResourceID = "doc-2" + if _, err := st.SaveProjectAuditLink(link); !errors.Is(err, ErrConflict) { + t.Fatalf("same key different binding: %v", err) + } + link.ResourceID = "doc-1" + link.IdempotencyKey = "outbox-2" + if _, err := st.SaveProjectAuditLink(link); !errors.Is(err, ErrConflict) { + t.Fatalf("same source different key: %v", err) + } + if err := st.EnsureUser("bob", "Bob"); err != nil { + t.Fatal(err) + } + bobConversation, err := st.CreateConversationInProject("bob", string(a.ID), "shared work", "") + if err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("bob", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "bob", CommandID: "rdm.confirm", OperationID: "op-bob", ConversationID: string(bobConversation.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + bobRows, _, _, err := st.ListProjectCommandAudits("bob", string(a.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + bobLink := ProjectAuditLink{HostAuditOffset: bobRows[0].Offset, ProjectID: string(a.ID), Owner: "bob", SourceKind: "domain-commit", SourceID: "commit-bob", OperationID: "op-bob", ResourceKind: "document", ResourceID: "doc-bob", DisclosureClass: "metadata", IdempotencyKey: "outbox-1"} + if _, err := st.SaveProjectAuditLink(bobLink); err != nil { + t.Fatalf("another owner may use the same outbox key: %v", err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "rdm.confirm", OperationID: "op-1", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + if _, err := st.CommittedAuditOffset("alice", string(a.ID), "op-1"); !errors.Is(err, ErrConflict) { + t.Fatalf("ambiguous host operation must not choose one: %v", err) + } +} diff --git a/internal/store/project_audit_sequence_test.go b/internal/store/project_audit_sequence_test.go new file mode 100644 index 0000000..c7d27fe --- /dev/null +++ b/internal/store/project_audit_sequence_test.go @@ -0,0 +1,610 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "os" + "sort" + "strings" + "sync" + "testing" + "time" + + agentcommand "agent-command" +) + +func openProjectAuditPG(t *testing.T) *Store { + t.Helper() + if os.Getenv("AGENT_PG_TEST_DSN") == "" { + t.Skip("AGENT_PG_TEST_DSN unset: isolated PostgreSQL check NOT_RUN") + } + st := openTestStore(t) + if st.Dialect() != "postgres" { + t.Fatal("expected PostgreSQL") + } + return st +} + +func TestProjectAuditAllWritersRejectUnsafeOnlineChanges(t *testing.T) { + st := openProjectAuditPG(t) + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "first", State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + row := `INSERT INTO command_audit(id,time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload) + VALUES ($1,'2026-10-14T00:00:00Z','explicit','','','committed','','','alice','{}')` + if _, err := st.db.Exec(row, 100); err == nil { + t.Fatal("online explicit ID above reserved sequence position was accepted") + } + if _, err := st.db.Exec(`UPDATE command_audit SET command_id='changed' WHERE id=1`); err == nil { + t.Fatal("online audit UPDATE was accepted") + } + if _, err := st.db.Exec(`DELETE FROM command_audit WHERE id=1`); err == nil { + t.Fatal("online audit DELETE was accepted") + } + for _, level := range []sql.IsolationLevel{sql.LevelRepeatableRead, sql.LevelSerializable} { + tx, err := st.db.BeginTx(context.Background(), &sql.TxOptions{Isolation: level}) + if err != nil { + t.Fatal(err) + } + _, err = tx.Exec(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload) + VALUES ('2026-10-14T00:00:00Z','stale-snapshot','','','committed','','','alice','{}')`) + tx.Rollback() + if err == nil || !strings.Contains(err.Error(), "read committed") { + t.Fatalf("isolation %v did not fail closed: %v", level, err) + } + } + seq := commandAuditSequenceName(t, st) + if _, err := st.db.Exec(`SELECT nextval($1::regclass)`, seq); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`INSERT INTO command_audit(id,time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload) + VALUES (4,'2026-10-14T00:00:00Z','first-of-batch','','','committed','','','alice','{}'), + (1,'2026-10-14T00:00:00Z','decreasing','','','committed','','','alice','{}')`); err == nil { + t.Fatal("unordered multirow audit INSERT was accepted") + } + if _, err := st.db.Exec(row+` ON CONFLICT DO NOTHING`, 1); err == nil { + t.Fatal("online replay bypassed the INSERT guard through ON CONFLICT") + } + var count int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM command_audit`).Scan(&count); err != nil || count != 1 { + t.Fatalf("unsafe writes persisted: count=%d err=%v", count, err) + } +} + +func TestProjectAuditAllWritersGuardDriftFailsClosed(t *testing.T) { + st := openProjectAuditPG(t) + for _, trigger := range []string{"project_audit_insert_guard_v1", "project_audit_mutation_guard_v1"} { + if _, err := st.db.Exec(`ALTER TABLE command_audit DISABLE TRIGGER ` + trigger); err != nil { + t.Fatal(err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil { + t.Fatalf("read accepted disabled trigger %s", trigger) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "drift"}); err == nil { + t.Fatalf("append accepted disabled trigger %s", trigger) + } + if _, err := st.db.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER ` + trigger); err != nil { + t.Fatal(err) + } + } + if _, err := st.db.Exec(`CREATE OR REPLACE TRIGGER project_audit_insert_guard_v1 BEFORE INSERT ON command_audit FOR EACH ROW EXECUTE FUNCTION project_audit_mutation_guard_v1()`); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_insert_guard_v1`); err != nil { + t.Fatal(err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil { + t.Fatal("read accepted rebound insert trigger") + } + if _, err := st.db.Exec(`CREATE OR REPLACE TRIGGER project_audit_insert_guard_v1 BEFORE INSERT ON command_audit FOR EACH ROW EXECUTE FUNCTION project_audit_insert_guard_v1()`); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`ALTER TABLE command_audit ENABLE ALWAYS TRIGGER project_audit_insert_guard_v1`); err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "restored"}); err != nil { + t.Fatalf("restored guard: %v", err) + } +} + +func commandAuditSequenceName(t *testing.T, st *Store) string { + t.Helper() + var name string + if err := st.db.QueryRow(`SELECT pg_get_serial_sequence('command_audit','id')`).Scan(&name); err != nil { + t.Fatal(err) + } + if name == "" { + t.Fatal("command_audit has no owned sequence") + } + return name +} + +func TestProjectAuditSequenceMigrationRejectsCacheTen(t *testing.T) { + st := openProjectAuditPG(t) + if err := st.MigrateDown(38); err != nil { + t.Fatal(err) + } + seq := commandAuditSequenceName(t, st) + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 10"); err != nil { + t.Fatal(err) + } + if err := st.migrate(); err == nil { + t.Fatal("migration accepted CACHE 10, which permits out-of-order IDs") + } + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 1"); err != nil { + t.Fatal(err) + } + if err := st.migrate(); err != nil { + t.Fatalf("safe CACHE 1 migration: %v", err) + } +} + +func TestProjectAuditSequenceRuntimeRejectsConfigurationDrift(t *testing.T) { + st := openProjectAuditPG(t) + seq := commandAuditSequenceName(t, st) + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 10"); err != nil { + t.Fatal(err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil { + t.Fatal("project audit reported a cutoff with CACHE 10") + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "unsafe-cache"}); err == nil { + t.Fatal("append allocated an ID with CACHE 10") + } + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 1"); err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "safe-cache"}); err != nil { + t.Fatalf("safe append after reset: %v", err) + } + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CYCLE"); err != nil { + t.Fatal(err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil { + t.Fatal("read accepted cyclic audit IDs") + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "cyclic"}); err == nil { + t.Fatal("append accepted cyclic audit IDs") + } + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " NO CYCLE"); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " MINVALUE -10"); err != nil { + t.Fatal(err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil { + t.Fatal("read accepted a nonpositive audit ID range") + } + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " MINVALUE 1"); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT nextval(pg_get_serial_sequence('command_audit','id')::regclass)`); err != nil { + t.Fatal(err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil { + t.Fatal("read accepted an unguarded sequence default") + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "unguarded-default"}); err == nil { + t.Fatal("append accepted an unguarded sequence default") + } + if _, err := st.db.Exec(`ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT CASE WHEN false THEN next_command_audit_id('command_audit'::regclass) ELSE nextval(pg_get_serial_sequence('command_audit','id')::regclass) END`); err != nil { + t.Fatal(err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", "project", 0, 0, 10); err == nil { + t.Fatal("read accepted a replaced default that merely mentions the guarded function") + } + if _, err := st.db.Exec(`ALTER TABLE command_audit ALTER COLUMN id SET DEFAULT next_command_audit_id('command_audit'::regclass)`); err != nil { + t.Fatal(err) + } +} + +func TestProjectAuditSequenceRejectsRewindIntoUnusedGap(t *testing.T) { + st := openProjectAuditPG(t) + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + appendAudit := func(id string) error { + return st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: id, ConversationID: string(c.ID), State: agentcommand.StateCommitted}) + } + if err := appendAudit("first"); err != nil { + t.Fatal(err) + } + seq := commandAuditSequenceName(t, st) + if _, err := st.db.Exec("SELECT nextval('" + seq + "'::regclass)"); err != nil { + t.Fatal(err) + } // consume one ID without a row + if err := appendAudit("third"); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec("SELECT setval('" + seq + "'::regclass,1)"); err != nil { + t.Fatal(err) + } + if err := appendAudit("rewound-into-gap"); err == nil { + t.Fatal("rewound sequence inserted a low ID into a committed-prefix gap") + } + if _, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10); err == nil { + t.Fatal("read claimed a safe cutoff after sequence rewind") + } + if _, err := st.db.Exec("SELECT setval('" + seq + "'::regclass,3)"); err != nil { + t.Fatal(err) + } + if err := appendAudit("recovered"); err != nil { + t.Fatalf("append after restoring sequence: %v", err) + } +} + +// Advisory locking alone does not fix cached sequence allocation: session A +// can hold 2..10 while session B commits 11, then A later commits 2. +func TestProjectAuditSequenceCacheTenDefeatsAdvisoryLock(t *testing.T) { + st := openProjectAuditPG(t) + for _, query := range []string{ + `CREATE TABLE cache_probe(id BIGSERIAL PRIMARY KEY)`, + `ALTER SEQUENCE cache_probe_id_seq CACHE 10`, + `CREATE FUNCTION cache_probe_next_id(target regclass) RETURNS bigint LANGUAGE plpgsql AS $$ + BEGIN PERFORM pg_advisory_xact_lock(715328491,target::oid::int4); + RETURN nextval(pg_get_serial_sequence(target::text,'id')::regclass); END $$`, + `ALTER TABLE cache_probe ALTER COLUMN id SET DEFAULT cache_probe_next_id('cache_probe'::regclass)`, + } { + if _, err := st.db.Exec(query); err != nil { + t.Fatal(err) + } + } + ctx := context.Background() + a, err := st.db.Conn(ctx) + if err != nil { + t.Fatal(err) + } + defer a.Close() + b, err := st.db.Conn(ctx) + if err != nil { + t.Fatal(err) + } + defer b.Close() + var first, other, late int64 + if err := a.QueryRowContext(ctx, `INSERT INTO cache_probe DEFAULT VALUES RETURNING id`).Scan(&first); err != nil { + t.Fatal(err) + } + if err := b.QueryRowContext(ctx, `INSERT INTO cache_probe DEFAULT VALUES RETURNING id`).Scan(&other); err != nil { + t.Fatal(err) + } + if err := a.QueryRowContext(ctx, `INSERT INTO cache_probe DEFAULT VALUES RETURNING id`).Scan(&late); err != nil { + t.Fatal(err) + } + if first != 1 || other != 11 || late != 2 { + t.Fatalf("CACHE 10 counterexample changed: A=%d B=%d A-late=%d", first, other, late) + } +} + +func TestProjectAuditSequenceNeverCommitsPreUpgradeCachedLowID(t *testing.T) { + st := openProjectAuditPG(t) + if err := st.MigrateDown(38); err != nil { + t.Fatal(err) + } + seq := commandAuditSequenceName(t, st) + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 10"); err != nil { + t.Fatal(err) + } + ctx := context.Background() + a, err := st.db.Conn(ctx) + if err != nil { + t.Fatal(err) + } + defer a.Close() + b, err := st.db.Conn(ctx) + if err != nil { + t.Fatal(err) + } + defer b.Close() + insert := `INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload) + VALUES ('2026-10-14T00:00:00Z','legacy','','','committed','','','alice','{}') RETURNING id` + var first, high, late int64 + if err := a.QueryRowContext(ctx, insert).Scan(&first); err != nil { + t.Fatal(err) + } + if err := b.QueryRowContext(ctx, insert).Scan(&high); err != nil { + t.Fatal(err) + } + if first != 1 || high != 11 { + t.Fatalf("old cached allocation: first=%d high=%d", first, high) + } + if _, err := st.db.Exec("ALTER SEQUENCE " + seq + " CACHE 1"); err != nil { + t.Fatal(err) + } + if err := st.migrate(); err != nil { + t.Fatalf("upgrade with idle cached connections: %v", err) + } + if err := a.QueryRowContext(ctx, insert).Scan(&late); err == nil && late <= high { + t.Fatalf("old connection committed late low ID %d below %d", late, high) + } + var count int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM command_audit WHERE id=2`).Scan(&count); err != nil || count != 0 { + t.Fatalf("late cached row persisted: count=%d err=%v", count, err) + } +} + +func TestProjectAuditSequenceMigrationRoundTripPreservesLinks(t *testing.T) { + st := openProjectAuditPG(t) + if err := st.EnsureUser("alice", "Alice"); err != nil { + t.Fatal(err) + } + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "rdm.confirm", OperationID: "op-1", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + rows, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + offset := rows[0].Offset + link := ProjectAuditLink{HostAuditOffset: offset, ProjectID: string(p.ID), Owner: "alice", SourceKind: "domain-commit", SourceID: "commit-1", OperationID: "op-1", ResourceKind: "document", ResourceID: "doc-1", DisclosureClass: "metadata", IdempotencyKey: "outbox-1"} + if _, err := st.SaveProjectAuditLink(link); err != nil { + t.Fatal(err) + } + if err := st.MigrateDown(38); err != nil { + t.Fatalf("down to v38: %v", err) + } + if _, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10); err == nil { + t.Fatal("old default exposed a strong cutoff after down") + } + if _, row, err := st.ProjectAuditLinkedReceipt(string(p.ID), "domain-commit", "commit-1", offset); err != nil || row.Offset != offset { + t.Fatalf("link changed during down: %+v %v", row, err) + } + if err := st.migrate(); err != nil { + t.Fatalf("up to v39: %v", err) + } + rows, _, _, err = st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil || len(rows) != 1 || rows[0].Offset != offset { + t.Fatalf("restored audit: %+v %v", rows, err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "after-up", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + rows, _, _, err = st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil || len(rows) != 2 || rows[0].Offset <= offset { + t.Fatalf("sequence not advanced after up: %+v %v", rows, err) + } +} + +func TestProjectAuditSequenceLockTimeoutAndRollback(t *testing.T) { + st := openProjectAuditPG(t) + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + tx, err := st.db.Begin() + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + insert := st.q(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload,actor_name,project_id) + VALUES (?,?,'','','committed','',?,?,?,?,?) RETURNING id`) + args := []any{time.Now().UTC().Format(time.RFC3339Nano), "held", string(c.ID), "alice", `{}`, "Alice", string(p.ID)} + var heldID int64 + if err := tx.QueryRow(insert, args...).Scan(&heldID); err != nil { + t.Fatal(err) + } + ctx := context.Background() + conn, err := st.db.Conn(ctx) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err := conn.ExecContext(ctx, `SET lock_timeout='150ms'`); err != nil { + t.Fatal(err) + } + var blockedID int64 + err = conn.QueryRowContext(ctx, insert, args...).Scan(&blockedID) + if err == nil || !strings.Contains(err.Error(), "lock timeout") { + t.Fatalf("second writer did not stop on advisory lock: id=%d err=%v", blockedID, err) + } + if _, err := conn.ExecContext(ctx, `RESET lock_timeout`); err != nil { + t.Fatal(err) + } + cancelCtx, cancel := context.WithTimeout(ctx, 150*time.Millisecond) + defer cancel() + var canceledID int64 + if err := conn.QueryRowContext(cancelCtx, insert, args...).Scan(&canceledID); err == nil { + t.Fatalf("canceled writer passed advisory lock: id=%d", canceledID) + } + if err := tx.Rollback(); err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "after-rollback", ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + rows, cutoff, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil || len(rows) != 1 || rows[0].Offset <= heldID || cutoff != rows[0].Offset { + t.Fatalf("rollback gap affected cutoff: rows=%+v cutoff=%d held=%d err=%v", rows, cutoff, heldID, err) + } +} + +func TestProjectAuditSequenceConcurrentLoad(t *testing.T) { + st := openProjectAuditPG(t) + var serverVersion string + if err := st.db.QueryRow(`SHOW server_version`).Scan(&serverVersion); err != nil { + t.Fatal(err) + } + if err := st.EnsureUser("alice", "Alice"); err != nil { + t.Fatal(err) + } + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + const writers, each = 8, 24 + run := func(label string, write func(agentcommand.AuditEnvelope) error) (time.Duration, time.Duration) { + t.Helper() + durations := make([]time.Duration, writers*each) + errors := make(chan error, writers*each) + start := time.Now() + var wg sync.WaitGroup + for worker := 0; worker < writers; worker++ { + wg.Add(1) + go func(worker int) { + defer wg.Done() + for i := 0; i < each; i++ { + env := agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: fmt.Sprintf("%s-%d-%d", label, worker, i), ConversationID: string(c.ID), State: agentcommand.StateCommitted} + at := time.Now() + if err := write(env); err != nil { + errors <- err + return + } + durations[worker*each+i] = time.Since(at) + } + }(worker) + } + wg.Wait() + close(errors) + for err := range errors { + t.Fatalf("%s append: %v", label, err) + } + elapsed := time.Since(start) + sort.Slice(durations, func(i, j int) bool { return durations[i] < durations[j] }) + return elapsed, durations[len(durations)*95/100] + } + if err := st.MigrateDown(38); err != nil { + t.Fatal(err) + } + oldQuery := st.q(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload,actor_name,project_id) + VALUES (?,?,?,?,?,?,?,?,?,COALESCE((SELECT name FROM users WHERE owner=?),''),COALESCE((SELECT project_id FROM conversations WHERE id=? AND owner=?),''))`) + oldElapsed, oldP95 := run("old", func(env agentcommand.AuditEnvelope) error { + raw, err := json.Marshal(env) + if err != nil { + return err + } + _, err = st.db.Exec(oldQuery, env.Time.UTC().Format(time.RFC3339Nano), env.CommandID, "", "", string(env.State), "", env.ConversationID, "alice", string(raw), env.Actor, env.ConversationID, "alice") + return err + }) + if err := st.migrate(); err != nil { + t.Fatalf("upgrade after baseline: %v", err) + } + statusMode := "MAX-first unlocked status" + if os.Getenv("AGENT_AUDIT_BENCH_READ_LOCK") == "1" { + // A test-only wrapper reconstructs the earlier status-read lock so the + // isolated load receipt can compare its cost against the final design. + if _, err := st.db.Exec(`ALTER FUNCTION project_audit_cutoff_safe(regclass) RENAME TO project_audit_cutoff_unlocked`); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`CREATE FUNCTION project_audit_cutoff_safe(target regclass) RETURNS boolean LANGUAGE plpgsql VOLATILE AS $$ + BEGIN PERFORM pg_advisory_xact_lock(715328491,target::oid::int4); + RETURN project_audit_cutoff_unlocked(target); END $$`); err != nil { + t.Fatal(err) + } + statusMode = "test-only locked status wrapper" + } + newElapsed, newP95 := run("guarded", func(env agentcommand.AuditEnvelope) error { return st.AppendAudit("alice", env) }) + t.Logf("isolated PostgreSQL %s audit append: %d writes per mode / %d workers; v38 direct INSERT=%s p95=%s (%.0f/s), v40 Store.AppendAudit with %s=%s p95=%s (%.0f/s); hardware/load dependent SELF only", + serverVersion, writers*each, writers, oldElapsed, oldP95, float64(writers*each)/oldElapsed.Seconds(), statusMode, newElapsed, newP95, float64(writers*each)/newElapsed.Seconds()) + seen := map[string]bool{} + var before, cutoff int64 + for { + page, fixed, next, err := st.ListProjectCommandAudits("alice", string(p.ID), before, cutoff, 100) + if err != nil { + t.Fatal(err) + } + if cutoff == 0 { + cutoff = fixed + } + for _, row := range page { + if seen[row.Envelope.CommandID] { + t.Fatalf("duplicate paged command: %s", row.Envelope.CommandID) + } + seen[row.Envelope.CommandID] = true + } + if next == 0 { + break + } + before = next + } + if len(seen) != 2*writers*each { + t.Fatalf("paged commands=%d want=%d", len(seen), 2*writers*each) + } +} + +func TestProjectAuditConcurrentReadWriteNeverReportsFalseDrift(t *testing.T) { + st := openProjectAuditPG(t) + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + for i := 0; i < 210; i++ { + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: fmt.Sprintf("seed-%d", i), ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + } + const writers, writes, readers, reads = 6, 30, 4, 40 + errs := make(chan error, writers*writes+readers*reads) + readDurations := make(chan time.Duration, readers*reads) + var wg sync.WaitGroup + for worker := 0; worker < writers; worker++ { + wg.Add(1) + go func(worker int) { + defer wg.Done() + for i := 0; i < writes; i++ { + err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: fmt.Sprintf("write-%d-%d", worker, i), ConversationID: string(c.ID), State: agentcommand.StateCommitted}) + if err != nil { + errs <- fmt.Errorf("append: %w", err) + return + } + } + }(worker) + } + for reader := 0; reader < readers; reader++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < reads; i++ { + started := time.Now() + page, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 200) + if err != nil { + errs <- fmt.Errorf("read: %w", err) + return + } + if len(page) != 200 { + errs <- fmt.Errorf("large page rows=%d want=200", len(page)) + return + } + readDurations <- time.Since(started) + } + }() + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatal(err) + } + close(readDurations) + var samples []time.Duration + for duration := range readDurations { + samples = append(samples, duration) + } + sort.Slice(samples, func(i, j int) bool { return samples[i] < samples[j] }) + t.Logf("isolated PG mixed load: %d append writes / %d writers and %d 200-row pages / %d readers; page p95=%s, SELF cost signal only", + writers*writes, writers, readers*reads, readers, samples[len(samples)*95/100]) +} diff --git a/internal/store/project_audit_test.go b/internal/store/project_audit_test.go new file mode 100644 index 0000000..7234dd7 --- /dev/null +++ b/internal/store/project_audit_test.go @@ -0,0 +1,267 @@ +package store + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + agentcommand "agent-command" +) + +func TestProjectCommandAuditsUseStoredConversationProjectAndFixedCutoff(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "audit.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnsureUser("alice", "Alice at action time"); err != nil { + t.Fatal(err) + } + projectA, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + projectB, err := st.CreateProject("alice", "B") + if err != nil { + t.Fatal(err) + } + convA, err := st.CreateConversation("alice", string(projectA.ID), "A work", "") + if err != nil { + t.Fatal(err) + } + convB, err := st.CreateConversation("alice", string(projectB.ID), "B work", "") + if err != nil { + t.Fatal(err) + } + appendCommand := func(owner, conversation, command string) { + t.Helper() + if err := st.AppendAudit(owner, agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: command, ConversationID: conversation, State: agentcommand.StateCommitted, ParamsDigest: "private-digest"}); err != nil { + t.Fatal(err) + } + } + appendCommand("alice", string(convA.ID), "a.one") + appendCommand("alice", string(convB.ID), "b.one") + appendCommand("alice", "client-supplied-only", "unbound") + appendCommand("bob", string(convA.ID), "wrong-owner") + appendCommand("alice", string(convA.ID), "a.two") + var expectedCutoff int64 + if err := st.db.QueryRow(st.q(`SELECT id FROM command_audit WHERE command_id=?`), "a.two").Scan(&expectedCutoff); err != nil { + t.Fatal(err) + } + appendCommand("alice", string(convB.ID), "b.latest") + + first, cutoff, next, err := st.ListProjectCommandAudits("alice", string(projectA.ID), 0, 0, 1) + if err != nil { + t.Fatal(err) + } + if len(first) != 1 || first[0].Envelope.CommandID != "a.two" || first[0].ActorName != "Alice at action time" || next == 0 || cutoff != expectedCutoff { + t.Fatalf("first page: rows=%+v cutoff=%d next=%d", first, cutoff, next) + } + appendCommand("alice", string(convA.ID), "a.late") + second, fixed, next, err := st.ListProjectCommandAudits("alice", string(projectA.ID), next, cutoff, 1) + if err != nil { + t.Fatal(err) + } + if fixed != cutoff || len(second) != 1 || second[0].Envelope.CommandID != "a.one" || next != 0 { + t.Fatalf("second page: rows=%+v cutoff=%d next=%d", second, fixed, next) + } + other, _, _, err := st.ListProjectCommandAudits("alice", string(projectB.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + if len(other) != 2 || other[0].Envelope.CommandID != "b.latest" || other[1].Envelope.CommandID != "b.one" { + t.Fatalf("B rows: %+v", other) + } + unknown, _, _, err := st.ListProjectCommandAudits("alice", "untrusted-project", 0, 0, 10) + if err != nil { + t.Fatal(err) + } + if len(unknown) != 0 { + t.Fatalf("untrusted project rows: %+v", unknown) + } +} + +func TestProjectCommandAuditKeepsActorNameSnapshot(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "audit.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnsureUser("alice", "Former name"); err != nil { + t.Fatal(err) + } + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "A work", "") + if err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "old", ConversationID: string(c.ID)}); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(st.q(`UPDATE users SET name=? WHERE owner=?`), "Current name", "alice"); err != nil { + t.Fatal(err) + } + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "new", ConversationID: string(c.ID)}); err != nil { + t.Fatal(err) + } + legacy, err := json.Marshal(agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "legacy", ConversationID: string(c.ID)}) + if err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(st.q(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, owner, payload) VALUES (?, ?, '', '', '', '', ?, ?, ?)`), time.Now().UTC().Format(time.RFC3339Nano), "legacy", string(c.ID), "alice", string(legacy)); err != nil { + t.Fatal(err) + } + rows, _, _, err := st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + if len(rows) != 2 || rows[0].ActorName != "Current name" || rows[1].ActorName != "Former name" { + t.Fatalf("names: %+v", rows) + } + var legacyProject string + if err := st.db.QueryRow(st.q(`SELECT project_id FROM command_audit WHERE command_id=?`), "legacy").Scan(&legacyProject); err != nil { + t.Fatal(err) + } + if legacyProject != "" { + t.Fatalf("legacy attribution was fabricated: %q", legacyProject) + } +} + +func TestProjectCommandAuditDoesNotMoveWithConversation(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "audit.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + a, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + b, err := st.CreateProject("alice", "B") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(a.ID), "work", "") + if err != nil { + t.Fatal(err) + } + appendCommand := func(id string) { + t.Helper() + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: id, ConversationID: string(c.ID)}); err != nil { + t.Fatal(err) + } + } + appendCommand("before-move") + if _, err := st.db.Exec(st.q(`UPDATE conversations SET project_id=? WHERE id=?`), string(b.ID), string(c.ID)); err != nil { + t.Fatal(err) + } + appendCommand("after-move") + oldProject, _, _, err := st.ListProjectCommandAudits("alice", string(a.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + newProject, _, _, err := st.ListProjectCommandAudits("alice", string(b.ID), 0, 0, 10) + if err != nil { + t.Fatal(err) + } + if len(oldProject) != 1 || oldProject[0].Envelope.CommandID != "before-move" || len(newProject) != 1 || newProject[0].Envelope.CommandID != "after-move" { + t.Fatalf("project attribution changed: A=%+v B=%+v", oldProject, newProject) + } +} + +// Under the old PostgreSQL SERIAL default, an uncommitted middle ID can be +// skipped after a higher committed row advances the page cursor past it. +// The committed-prefix implementation must keep the later writer waiting. +func TestProjectAuditPostgresCursorNeverSkipsLateCommit(t *testing.T) { + if os.Getenv("AGENT_PG_TEST_DSN") == "" { + t.Skip("AGENT_PG_TEST_DSN unset: PG concurrency check NOT_RUN") + } + st := openTestStore(t) + if st.Dialect() != "postgres" { + t.Fatal("expected postgres") + } + if err := st.EnsureUser("alice", "Alice"); err != nil { + t.Fatal(err) + } + p, err := st.CreateProject("alice", "A") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "work", "") + if err != nil { + t.Fatal(err) + } + for _, command := range []string{"lower-one", "lower-two"} { + if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: command, ConversationID: string(c.ID), State: agentcommand.StateCommitted}); err != nil { + t.Fatal(err) + } + } + env := agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "middle-pending", ConversationID: string(c.ID), State: agentcommand.StateCommitted} + raw, err := json.Marshal(env) + if err != nil { + t.Fatal(err) + } + tx, err := st.db.Begin() + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + var middleID int64 + if err := tx.QueryRow(st.q(`INSERT INTO command_audit(time,command_id,source,entrance,state,operation_id,conversation_id,owner,payload,actor_name,project_id) + VALUES (?,?,'','','committed','',?,?,?,?,?) RETURNING id`), time.Now().UTC().Format(time.RFC3339Nano), "middle-pending", string(c.ID), "alice", string(raw), "Alice", string(p.ID)).Scan(&middleID); err != nil { + t.Fatal(err) + } + highDone := make(chan error, 1) + started := make(chan struct{}) + go func() { + close(started) + highDone <- st.AppendAudit("alice", agentcommand.AuditEnvelope{Time: time.Now(), Actor: "alice", CommandID: "higher-committed", ConversationID: string(c.ID), State: agentcommand.StateCommitted}) + }() + <-started + highCommittedEarly := false + select { + case err := <-highDone: + if err != nil { + t.Fatal(err) + } + highCommittedEarly = true + case <-time.After(250 * time.Millisecond): + } + var first []ProjectCommandAuditRow + var cutoff, next int64 + if highCommittedEarly { + first, cutoff, next, err = st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 2) + if err != nil { + tx.Rollback() + t.Fatal(err) + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + if !highCommittedEarly { + if err := <-highDone; err != nil { + t.Fatal(err) + } + first, cutoff, next, err = st.ListProjectCommandAudits("alice", string(p.ID), 0, 0, 2) + if err != nil { + t.Fatal(err) + } + } + second, fixed, _, err := st.ListProjectCommandAudits("alice", string(p.ID), next, cutoff, 2) + if err != nil { + t.Fatal(err) + } + seen := map[string]bool{} + for _, row := range append(first, second...) { + seen[row.Envelope.CommandID] = true + } + if next == 0 || fixed != cutoff || len(seen) != 4 || !seen["middle-pending"] { + t.Fatalf("cross-page audit omitted a committed row: early=%v middleID=%d first=%+v second=%+v cutoff=%d next=%d", highCommittedEarly, middleID, first, second, cutoff, next) + } +} diff --git a/pkg/extension/extension.go b/pkg/extension/extension.go index f36e91e..5020234 100644 --- a/pkg/extension/extension.go +++ b/pkg/extension/extension.go @@ -34,7 +34,21 @@ import ( // 分区组件自身的命令面 fail-closed）。 // 2.12.0-experimental: optional ProjectHost/ProjectLifecycleRegistry and trusted // ResourceAuthzQuery.Source for per-document publication authorization. -const SurfaceVersion = "2.13.0-experimental" +// 2.13.0-experimental: optional ProjectAuditHost and ProjectAuditRegistry (X2 +// project-bound command-audit read slice and domain audit.read callback). +// 2.14.0-experimental: X2 optional committed project-audit link publisher, +// verifier, disclosure callback, and exact linked receipt read. +// 2.15.0-experimental: trusted Request.OperationID before handler execution and +// retryable ProjectAuditLinkRequest.HostAuditOffset=0 resolution. +// 2.16.0-experimental: cryptographic operation IDs and registration-bound +// ProjectAuditPublisherFactory; scoped Host string IDs cannot publish links. +// 2.17.0-experimental: PostgreSQL project audit pages require a verified +// committed-prefix ID generator; unsafe sequence/default drift rejects reads. +// 2.18.0-experimental: all PostgreSQL audit writes require the database +// insert guard and append-only policy before a committed-prefix claim. +// 2.19.0-experimental: combine optional X1 project context/source recheck and +// X2 project audit capabilities on one Backend surface. +const SurfaceVersion = "2.19.0-experimental" // ErrNotLinked 表示进程没有空白导入 link 包，Main 无法启动服务。 var ErrNotLinked = errors.New("extension: runner not linked; blank-import agent-system-backend/pkg/extension/link") diff --git a/pkg/extension/host.go b/pkg/extension/host.go index 9d36b71..df21509 100644 --- a/pkg/extension/host.go +++ b/pkg/extension/host.go @@ -14,6 +14,9 @@ type Request struct { Actor string Owner string ConversationID string + // OperationID is generated by the trusted host bus before this handler runs. + // Persist it with an actual domain commit/outbox; never treat it as a retry key. + OperationID string // HostProjectID 本次分发所属基座项目（EXT-18 可选上下文；空＝无项目 // 上下文——API 入口无对话或对话未关联项目时为零值）。下游按显式映射 // 解析（如 RDM rdm_host_project_map），不得据此替代自身授权判定。 diff --git a/pkg/extension/project_audit.go b/pkg/extension/project_audit.go new file mode 100644 index 0000000..c05d6dc --- /dev/null +++ b/pkg/extension/project_audit.go @@ -0,0 +1,161 @@ +package extension + +import ( + "context" + "errors" + "time" + + core "agent-system-core" +) + +// ProjectAuditAuthorizer answers the domain's audit.read decision for the +// authenticated actor and a project identified by trusted host storage. +// A project ID supplied by a caller is a locator, never an authorization. +type ProjectAuditAuthorizer func(actor string, projectID core.ID) (bool, error) + +// ProjectAuditRegistry is optional so older extension registries remain valid. +// Only an in-process domain plugin can register this callback. +type ProjectAuditRegistry interface { + RegisterProjectAuditAuthorizer(ProjectAuditAuthorizer) error +} + +var ( + ErrProjectAuditDenied = errors.New("extension: project audit denied") + ErrProjectAuditUnsupported = errors.New("extension: project audit unsupported") + ErrProjectAuditLinkConflict = errors.New("extension: project audit link conflict") + ErrProjectAuditPending = errors.New("extension: committed host audit receipt pending") +) + +// ProjectAuditLinkRequest binds one host command receipt to an already +// committed domain fact. Owner is never caller supplied; the host handle and +// stored audit row establish it. HostAuditOffset=0 asks the host to resolve a +// unique committed row by its trusted owner/project/OperationID, which lets an +// outbox wait for the bus audit commit after the domain handler returns. This +// first slice supports domain-commit only. +type ProjectAuditLinkRequest struct { + ProjectID core.ID `json:"projectId"` + HostAuditOffset int64 `json:"hostAuditOffset"` + SourceKind string `json:"sourceKind"` + SourceID string `json:"sourceId"` + OperationID string `json:"operationId"` + ResourceKind string `json:"resourceKind"` + ResourceID string `json:"resourceId"` + RunID string `json:"runId,omitempty"` + DisclosureClass string `json:"disclosureClass"` + IdempotencyKey string `json:"idempotencyKey"` +} + +// ProjectAuditLinkProof contains the caller request plus identity and source +// fields re-read from trusted host storage. The domain verifier must check its +// committed source event and operation/resource attribution, not a model claim. +type ProjectAuditLinkProof struct { + ProjectAuditLinkRequest + Owner string +} + +type ProjectAuditLinkVerifier func(context.Context, ProjectAuditLinkProof) (bool, error) + +// ProjectAuditLinkedResource is the stored, immutable link presented to the +// current domain disclosure policy. A caller ID alone never authorizes it. +type ProjectAuditLinkedResource struct { + ProjectID core.ID + Owner string + SourceKind string + SourceID string + OperationID string + ResourceKind string + ResourceID string + RunID string + DisclosureClass string +} + +type ProjectAuditDisclosure struct { + Metadata bool + Body bool +} + +type ProjectAuditDisclosureAuthorizer func(context.Context, string, ProjectAuditLinkedResource) (ProjectAuditDisclosure, error) + +// ProjectAuditPublisherFactory is issued only by successful verifier +// registration. A domain outbox may retain it and bind an owner for each job; +// owner and committed-source attribution are checked again when publishing. +// NewScopedHost(owner, pluginID) cannot mint this capability. +type ProjectAuditPublisherFactory interface { + ForOwner(owner string) (ProjectAuditPublisher, error) +} + +// ProjectAuditLinkRegistry is optional. Only the registering compiled domain +// plugin receives a publisher factory; string plugin IDs carry no authority. +type ProjectAuditLinkRegistry interface { + RegisterProjectAuditLinkVerifier(ProjectAuditLinkVerifier) (ProjectAuditPublisherFactory, error) + RegisterProjectAuditDisclosureAuthorizer(ProjectAuditDisclosureAuthorizer) error +} + +type ProjectAuditLinkReceipt struct { + LinkID int64 `json:"linkId"` + HostAuditOffset int64 `json:"hostAuditOffset"` +} + +type ProjectAuditPublisher interface { + PublishProjectAuditLink(context.Context, ProjectAuditLinkRequest) (ProjectAuditLinkReceipt, error) +} + +// ProjectAuditQuery reads one project-bound command-audit slice. Offset is a +// host source offset, not a cross-domain event sequence. A nonzero BeforeOffset +// must be paired with the CutoffOffset returned on the first page. PostgreSQL +// pages are served only while the installed generator, sequence, and all-writer +// database guards prove a +// committed prefix; otherwise the host rejects the read instead of claiming a +// fixed snapshot. See Coverage.CutoffConsistency. +type ProjectAuditQuery struct { + ProjectID core.ID `json:"projectId"` + BeforeOffset int64 `json:"beforeOffset,omitempty"` + CutoffOffset int64 `json:"cutoffOffset,omitempty"` + Limit int `json:"limit,omitempty"` + // Linked* select one exact committed link. This avoids enumerating hidden + // private links or leaking their counts before cursor policy is complete. + LinkedSourceKind string `json:"linkedSourceKind,omitempty"` + LinkedSourceID string `json:"linkedSourceId,omitempty"` + LinkedHostOffset int64 `json:"linkedHostOffset,omitempty"` +} + +// HostAuditReceipt is a server-projected receipt, never a raw audit payload. +// ActorName is the name recorded with this command; old rows have no snapshot. +type HostAuditReceipt struct { + SourceOffset int64 `json:"sourceOffset"` + Time time.Time `json:"time"` + ActorID string `json:"actorId,omitempty"` + ActorName string `json:"actorName,omitempty"` + ActorNameStatus string `json:"actorNameStatus"` + CommandID string `json:"commandId"` + State string `json:"state"` + OperationID string `json:"operationId,omitempty"` + ConversationID string `json:"conversationId"` +} + +// ProjectAuditCoverage states the exact scope of this first X2 slice. Missing +// host sources and old unbound rows cannot be inferred from owner-wide audit. +type ProjectAuditCoverage struct { + CommandAudit string `json:"commandAudit"` + CutoffConsistency string `json:"cutoffConsistency"` + CrossMember string `json:"crossMember"` + LegacyProjectBinding string `json:"legacyProjectBinding"` + RunRelations string `json:"runRelations"` + ModelReceipts string `json:"modelReceipts"` + BackgroundReceipts string `json:"backgroundReceipts"` + ResourceBodies string `json:"resourceBodies"` +} + +type ProjectAuditPage struct { + Receipts []HostAuditReceipt `json:"receipts"` + NextOffset int64 `json:"nextOffset,omitempty"` + CutoffOffset int64 `json:"cutoffOffset"` + Coverage ProjectAuditCoverage `json:"coverage"` +} + +// ProjectAuditHost is an optional, actor-bound read capability. A consumer must +// treat a failed type assertion as unsupported; an authorized page still reports +// partial coverage explicitly. +type ProjectAuditHost interface { + QueryProjectAudit(context.Context, ProjectAuditQuery) (ProjectAuditPage, error) +} diff --git a/pkg/extension/surface.txt b/pkg/extension/surface.txt index e8d69ad..d2b0aa9 100644 --- a/pkg/extension/surface.txt +++ b/pkg/extension/surface.txt @@ -8,6 +8,10 @@ AdminProvider.IsAdmin AdminZoneSpec Compiled ErrNotLinked +ErrProjectAuditDenied +ErrProjectAuditLinkConflict +ErrProjectAuditPending +ErrProjectAuditUnsupported Handler Host Host.ArtifactContent @@ -25,6 +29,7 @@ Host.ParseDocument Host.PluginID Host.TaskCreate Host.TaskUpdate +HostAuditReceipt HostClient InProc IntentSpec @@ -32,6 +37,28 @@ Main NewHostClient NewScopedHost Options +ProjectAuditAuthorizer +ProjectAuditCoverage +ProjectAuditDisclosure +ProjectAuditDisclosureAuthorizer +ProjectAuditHost +ProjectAuditHost.QueryProjectAudit +ProjectAuditLinkProof +ProjectAuditLinkReceipt +ProjectAuditLinkRegistry +ProjectAuditLinkRegistry.RegisterProjectAuditDisclosureAuthorizer +ProjectAuditLinkRegistry.RegisterProjectAuditLinkVerifier +ProjectAuditLinkRequest +ProjectAuditLinkVerifier +ProjectAuditLinkedResource +ProjectAuditPage +ProjectAuditPublisher +ProjectAuditPublisher.PublishProjectAuditLink +ProjectAuditPublisherFactory +ProjectAuditPublisherFactory.ForOwner +ProjectAuditQuery +ProjectAuditRegistry +ProjectAuditRegistry.RegisterProjectAuditAuthorizer ProjectContextBundle ProjectContextDenied ProjectContextIncomplete