From 928da87a9383f7f14d2c3f39a65832a1853f9b58 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 16:03:21 +0000 Subject: [PATCH] Require analyzed source manifests before project actions Why: A durable observation receipt must not be mistaken for completed risk analysis or permission to execute a project action. Evidence: Exact-source preflight tests cover pending versus proceed, partial attachment, post-ACK source tampering, current authorization, and receipt idempotency; go test -p 1 ./... and go build ./... passed. Impact: Add a fail-closed host preflight helper and per-message manifest read, still unwired to the public product path pending recovery and four-route coverage. Refs: TT-N03, TT-R01; approved X3 design r1 --- internal/app/project_work_observation.go | 88 ++++++++++++++++++- internal/app/project_work_observation_test.go | 67 ++++++++++++++ internal/store/project_work_outbox.go | 31 ++++++- internal/store/project_work_outbox_test.go | 4 + pkg/extension/project_work.go | 1 + 5 files changed, 188 insertions(+), 3 deletions(-) diff --git a/internal/app/project_work_observation.go b/internal/app/project_work_observation.go index c6bca54..cce3a94 100644 --- a/internal/app/project_work_observation.go +++ b/internal/app/project_work_observation.go @@ -2,6 +2,8 @@ package app import ( "context" + "crypto/sha256" + "encoding/json" "errors" "fmt" "sync" @@ -74,8 +76,17 @@ func deliverProjectWorkObservations(ctx context.Context, st *store.Store, slot * if err != nil { return 0, err } + return deliverProjectWorkRows(ctx, st, provider, work, rows) +} + +func deliverProjectWorkRows(ctx context.Context, st *store.Store, provider *extension.ProjectWorkObserver, + work func(context.Context, string, core.ID) (bool, error), rows []store.ProjectWorkObservationRow, +) (int, error) { completed := 0 for _, row := range rows { + if row.State != "pending" { + continue + } if err := ctx.Err(); err != nil { return completed, err } @@ -92,7 +103,7 @@ func deliverProjectWorkObservations(ctx context.Context, st *store.Store, slot * ProjectID: core.ID(row.ProjectID), ConversationID: core.ID(row.ConversationID), RunID: core.ID(row.RunID), SourceKind: row.SourceKind, SourceID: row.SourceID, SourceVersion: row.SourceVersion, SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex, SpanStart: row.SpanStart, SpanEnd: row.SpanEnd, - Text: row.Text, Scope: row.Scope, Delivery: row.Delivery, ObservedAt: row.ObservedAt, + Text: row.Text, Scope: row.Scope, Delivery: row.Delivery, DeliveryReason: row.Reason, ObservedAt: row.ObservedAt, } result, err := provider.Observe(ctx, observation) if err != nil { @@ -112,3 +123,78 @@ func deliverProjectWorkObservations(ctx context.Context, st *store.Store, slot * } return completed, nil } + +// preflightProjectMessage delivers this committed message's pending parts +// inline, then asks RDM to prove that every exact source part was analyzed for +// the pending action. A durable intake ACK alone never permits the action. +func preflightProjectMessage(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot, + work func(context.Context, string, core.ID) (bool, error), messageID, action string, +) (extension.ProjectWorkDecision, error) { + pending := extension.ProjectWorkDecision{State: extension.ProjectWorkPending, Reason: "project source awaiting analysis"} + provider := slot.get() + if provider == nil || work == nil { + return pending, errors.New("extension: trusted project work observer unavailable") + } + rows, err := st.ProjectMessageObservations(messageID) + if err != nil { + return pending, err + } + if len(rows) == 0 { + return pending, errors.New("extension: committed project message has no source manifest") + } + if _, err := deliverProjectWorkRows(ctx, st, provider, work, rows); err != nil { + return pending, err + } + rows, err = st.ProjectMessageObservations(messageID) + if err != nil { + return pending, err + } + manifest := make([]extension.ProjectWorkSource, 0, len(rows)) + for _, row := range rows { + if row.State != "accepted" || row.Delivery != "complete" { + return pending, nil + } + manifest = append(manifest, extension.ProjectWorkSource{ + EventID: row.EventID, SourceKind: row.SourceKind, SourceID: row.SourceID, + SourceVersion: row.SourceVersion, SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex, + }) + } + first := rows[0] + for _, row := range rows[1:] { + if row.Owner != first.Owner || row.ProjectID != first.ProjectID || row.ConversationID != first.ConversationID { + return pending, errors.New("extension: mixed project source manifest") + } + } + conv, err := st.GetConversation(first.Owner, first.ConversationID) + if err != nil || string(conv.ProjectID) != first.ProjectID { + return pending, errors.New("extension: project binding changed before preflight") + } + allowed, err := work(ctx, first.Owner, conv.ProjectID) + if err != nil || !allowed { + return pending, errors.New("extension: current project work denied before preflight") + } + raw, err := json.Marshal(manifest) + if err != nil { + return pending, err + } + watermark := fmt.Sprintf("%x", sha256.Sum256(raw)) + decision, err := provider.Preflight(ctx, extension.ProjectWorkAction{ + Actor: first.Owner, Owner: first.Owner, ProjectID: conv.ProjectID, + ConversationID: conv.ID, RunID: core.ID(first.RunID), Action: action, + Manifest: manifest, SourceWatermark: watermark, + }) + if err != nil { + return pending, err + } + switch decision.State { + case extension.ProjectWorkProceed: + if decision.DecisionID == "" || decision.SourceWatermark != watermark { + return pending, errors.New("extension: project preflight decision did not bind current sources") + } + case extension.ProjectWorkClarify, extension.ProjectWorkPending, extension.ProjectWorkDenied: + // A non-proceed decision is a real result, never an implied approval. + default: + return pending, errors.New("extension: invalid project preflight decision") + } + return decision, nil +} diff --git a/internal/app/project_work_observation_test.go b/internal/app/project_work_observation_test.go index 079ab87..31e4fbb 100644 --- a/internal/app/project_work_observation_test.go +++ b/internal/app/project_work_observation_test.go @@ -73,6 +73,73 @@ func TestProjectWorkDeliveryRechecksCurrentAuthorityAndACKsOnce(t *testing.T) { } } +func TestProjectMessagePreflightRequiresAnalysisAndExactManifest(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "preflight.db")) + 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", "P") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation("alice", string(project.ID), "Work", "") + if err != nil { + t.Fatal(err) + } + msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "要求效率和成本都低于人工"}}} + if err := st.AppendMessageWithRequestAccountChecked("alice", false, &msg, "req-1"); err != nil { + t.Fatal(err) + } + ready, observations, decisions := false, 0, 0 + observer := workObserverForTest() + observer.Observe = func(context.Context, extension.WorkObservation) (extension.WorkObservationResult, error) { + observations++ + return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: "intake-1"}, nil + } + observer.Preflight = func(_ context.Context, action extension.ProjectWorkAction) (extension.ProjectWorkDecision, error) { + decisions++ + if action.Actor != "alice" || action.ProjectID != project.ID || action.ConversationID != conv.ID || action.RunID != "" || action.Action != "conversation.run" || len(action.Manifest) != 1 { + t.Fatalf("preflight did not receive exact trusted source manifest: %+v", action) + } + if !ready { + return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil + } + return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-1", SourceWatermark: action.SourceWatermark}, nil + } + slot := &projectWorkObserverSlot{provider: &observer} + work := func(_ context.Context, actor string, id core.ID) (bool, error) { + return actor == "alice" && id == project.ID, nil + } + decision, err := preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run") + if err != nil || decision.State != extension.ProjectWorkPending || observations != 1 || decisions != 1 { + t.Fatalf("durable intake was confused with completed analysis: %+v observations=%d decisions=%d err=%v", decision, observations, decisions, err) + } + ready = true + decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run") + if err != nil || decision.State != extension.ProjectWorkProceed || observations != 1 || decisions != 2 { + t.Fatalf("same source did not proceed after domain analysis: %+v observations=%d decisions=%d err=%v", decision, observations, decisions, err) + } + if _, err := st.DB().Exec(`UPDATE messages SET parts=? WHERE id=?`, `[{"type":"text","text":"changed"}]`, msg.ID); err != nil { + t.Fatal(err) + } + decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run") + if err != nil || decision.State != extension.ProjectWorkPending || decisions != 2 { + t.Fatalf("source changed after ACK but stale decision still passed: %+v decisions=%d err=%v", decision, decisions, err) + } + partial := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartArtifactRef, Artifact: &core.ArtifactRef{ArtifactID: "file-1"}}}} + if err := st.AppendMessageWithRequestAccountChecked("alice", false, &partial, "req-2"); err != nil { + t.Fatal(err) + } + decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(partial.ID), "conversation.run") + if err != nil || decision.State != extension.ProjectWorkPending || decisions != 2 { + t.Fatalf("partial attachment reached project action preflight: %+v decisions=%d err=%v", decision, decisions, err) + } +} + func TestProjectWorkObservationRegistrationIsTrustedAndUnique(t *testing.T) { slot := &projectWorkObserverSlot{} trusted := pluginRegistry{pluginID: "rdm", caps: hostcaps{workObserver: slot}} diff --git a/internal/store/project_work_outbox.go b/internal/store/project_work_outbox.go index 9f39c26..de1a6dd 100644 --- a/internal/store/project_work_outbox.go +++ b/internal/store/project_work_outbox.go @@ -81,11 +81,38 @@ func (s *Store) appendProjectMessageObservationsTx(tx *sql.Tx, m *core.Message) // It rechecks that the immutable source bytes still match the recorded hash. // Missing or changed source bytes become partial, never invented complete text. func (s *Store) PendingProjectWorkObservations(limit int) ([]ProjectWorkObservationRow, error) { + return s.projectWorkObservations("", limit) +} + +// ProjectMessageObservations returns every committed part of one message, +// including already-ACKed parts, for an action's exact source manifest. +func (s *Store) ProjectMessageObservations(messageID string) ([]ProjectWorkObservationRow, error) { + if messageID == "" { + return nil, errors.New("project work observation: empty message ID") + } + var count int + if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM project_work_outbox WHERE source_id=?`), messageID).Scan(&count); err != nil { + return nil, err + } + if count > 200 { + return nil, errors.New("project work observation: message parts exceed safe preflight limit") + } + return s.projectWorkObservations(messageID, 200) +} + +func (s *Store) projectWorkObservations(messageID string, limit int) ([]ProjectWorkObservationRow, error) { if limit <= 0 || limit > 200 { limit = 50 } - rows, err := s.db.Query(s.q(`SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id - FROM project_work_outbox WHERE state='pending' ORDER BY observed_at,event_id LIMIT ?`), limit) + query := `SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id + FROM project_work_outbox WHERE state='pending' ORDER BY observed_at,event_id LIMIT ?` + args := []any{limit} + if messageID != "" { + query = `SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id + FROM project_work_outbox WHERE source_id=? ORDER BY part_index LIMIT ?` + args = []any{messageID, limit} + } + rows, err := s.db.Query(s.q(query), args...) if err != nil { return nil, err } diff --git a/internal/store/project_work_outbox_test.go b/internal/store/project_work_outbox_test.go index 38c9b57..377b4f6 100644 --- a/internal/store/project_work_outbox_test.go +++ b/internal/store/project_work_outbox_test.go @@ -84,6 +84,10 @@ func TestProjectMessageOutboxFollowsCommittedMessage(t *testing.T) { if err := st.AckProjectWorkObservation(textPart.EventID, "receipt-1"); err != nil { t.Fatal(err) } + allParts, err := st.ProjectMessageObservations(string(msg.ID)) + if err != nil || len(allParts) != 2 { + t.Fatalf("ACK erased exact action source manifest: %+v %v", allParts, err) + } if err := st.AckProjectWorkObservation(textPart.EventID, "receipt-1"); err != nil { t.Fatalf("same committed receipt was not idempotent: %v", err) } diff --git a/pkg/extension/project_work.go b/pkg/extension/project_work.go index 7aea802..85c66c6 100644 --- a/pkg/extension/project_work.go +++ b/pkg/extension/project_work.go @@ -29,6 +29,7 @@ type WorkObservation struct { Text string `json:"text,omitempty"` // only the actual authorized fragment Scope string `json:"scope"` // private|project|controlled Delivery string `json:"delivery"` // complete|summary-only|partial + DeliveryReason string `json:"deliveryReason,omitempty"` AuthorizationEpoch string `json:"authorizationEpoch"` ObservedAt time.Time `json:"observedAt"` }