diff --git a/internal/app/project_work_observation.go b/internal/app/project_work_observation.go index 3fb56dc..9c76b20 100644 --- a/internal/app/project_work_observation.go +++ b/internal/app/project_work_observation.go @@ -135,12 +135,18 @@ func deliverProjectWorkRows(ctx context.Context, st *store.Store, provider *exte if err != nil || !allowed { return completed, fmt.Errorf("project work source %s: current project authorization denied", row.EventID) } + account, err := st.GetAccountByOwner(row.Owner) + if err != nil || !account.Enabled { + return completed, fmt.Errorf("project work source %s: current account authorization unavailable", row.EventID) + } + authorizationEpoch := fmt.Sprintf("account:%d:project:%s", account.AuthVersion, row.ProjectID) observation := extension.WorkObservation{ EventID: row.EventID, Phase: "source-committed", Actor: row.Owner, Owner: row.Owner, 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, DeliveryReason: row.Reason, Metadata: row.Metadata, ObservedAt: row.ObservedAt, + Text: row.Text, Scope: row.Scope, Delivery: row.Delivery, DeliveryReason: row.Reason, Metadata: row.Metadata, + AuthorizationEpoch: authorizationEpoch, ObservedAt: row.ObservedAt, } result, err := provider.Observe(ctx, observation) if err != nil { @@ -322,6 +328,11 @@ func preflightProjectRun(ctx context.Context, st *store.Store, slot *projectWork if row.RunID != runID || row.Owner != string(run.Owner) || row.ProjectID != string(run.ProjectID) || row.ConversationID != string(run.ConversationID) { return pending, errors.New("extension: mixed run source manifest") } + if row.SourceKind != "user-message" && row.SourceKind != "assistant-message" { + if err := verify(ctx, row); err != nil { + return pending, err + } + } manifest = append(manifest, extension.ProjectWorkSource{ EventID: row.EventID, SourceKind: row.SourceKind, SourceID: row.SourceID, SourceVersion: row.SourceVersion, SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex, diff --git a/internal/app/project_work_observation_test.go b/internal/app/project_work_observation_test.go index 7e94bd5..3c48d3c 100644 --- a/internal/app/project_work_observation_test.go +++ b/internal/app/project_work_observation_test.go @@ -271,10 +271,13 @@ func TestRunPreflightWaitsForAllCommittedSourceParts(t *testing.T) { SourceVersion: "v1", Text: "工具中的测量结果", Delivery: "complete", Metadata: map[string]string{"sourceStatus": "transient", "toolName": "measure"}}); err != nil { t.Fatal(err) } - ready, preflights, observed := false, 0, 0 + ready, preflights, observed, verified := false, 0, 0, 0 provider := workObserverForTest() - provider.Observe = func(context.Context, extension.WorkObservation) (extension.WorkObservationResult, error) { + provider.Observe = func(_ context.Context, observation extension.WorkObservation) (extension.WorkObservationResult, error) { observed++ + if observation.AuthorizationEpoch == "" { + t.Fatal("observation omitted the current authorization epoch") + } return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: fmt.Sprintf("receipt-%d", observed)}, nil } provider.Preflight = func(_ context.Context, action extension.ProjectWorkAction) (extension.ProjectWorkDecision, error) { @@ -290,16 +293,17 @@ func TestRunPreflightWaitsForAllCommittedSourceParts(t *testing.T) { slot := &projectWorkObserverSlot{provider: &provider} work := func(context.Context, string, core.ID) (bool, error) { return true, nil } verify := func(ctx context.Context, row store.ProjectWorkObservationRow) error { + verified++ return verifyProjectTransientObservation(ctx, st, row) } decision, err := preflightProjectRun(context.Background(), st, slot, work, verify, string(run.ID), "model.consume.tool") - if err != nil || decision.State != extension.ProjectWorkPending || observed != 2 || preflights != 1 { + if err != nil || decision.State != extension.ProjectWorkPending || observed != 2 || preflights != 1 || verified != 2 { t.Fatalf("intake ACK was mistaken for finished risk analysis: %+v observed=%d preflights=%d err=%v", decision, observed, preflights, err) } ready = true decision, err = preflightProjectRun(context.Background(), st, slot, work, verify, string(run.ID), "model.consume.tool") - if err != nil || decision.State != extension.ProjectWorkProceed || observed != 2 || preflights != 2 { - t.Fatalf("fully analyzed run was not released once: %+v observed=%d preflights=%d err=%v", decision, observed, preflights, err) + if err != nil || decision.State != extension.ProjectWorkProceed || observed != 2 || preflights != 2 || verified != 3 { + t.Fatalf("fully analyzed run was not released once: %+v observed=%d verified=%d preflights=%d err=%v", decision, observed, verified, preflights, err) } } diff --git a/internal/runmgr/project_work_gate.go b/internal/runmgr/project_work_gate.go index 865aa55..1e83595 100644 --- a/internal/runmgr/project_work_gate.go +++ b/internal/runmgr/project_work_gate.go @@ -91,7 +91,7 @@ func resolveProjectWorkDecision(ctx context.Context, actor, projectID, runID, ac } response, err := ask(core.HITLCard{ Question: decision.Question, Kind: core.HITLKindChoice, Choices: choices, - AllowFreeInput: true, FreeInputHint: "也可以输入更准确的表述", Blocked: true, + AllowFreeInput: false, Blocked: true, BlockReason: decision.Reason, }) if err != nil { @@ -120,5 +120,15 @@ func resolveProjectWorkDecision(ctx context.Context, actor, projectID, runID, ac if resolved.State != extension.ProjectWorkProceed || resolved.DecisionID == "" { return pending, errors.New("project work clarification did not commit a proceed decision") } - return resolved, nil + refreshed, err := preflight(ctx, runID, action) + if err != nil { + return pending, err + } + if refreshed.State != extension.ProjectWorkProceed || refreshed.DecisionID == "" { + return pending, errors.New("project work sources changed or remain blocked after clarification") + } + if decision.SourceWatermark != "" && refreshed.SourceWatermark != decision.SourceWatermark { + return pending, errors.New("project work source watermark changed while clarification was pending") + } + return refreshed, nil } diff --git a/internal/runmgr/project_work_gate_test.go b/internal/runmgr/project_work_gate_test.go index bee3012..40f7e63 100644 --- a/internal/runmgr/project_work_gate_test.go +++ b/internal/runmgr/project_work_gate_test.go @@ -49,7 +49,7 @@ func TestProjectWorkGateClarifiesBeforeProceeding(t *testing.T) { Question: "效率低于人工是指更慢吗?", Options: []string{"效率更高且成本更低", "完成时间和成本更低", "确实允许效率更低"}, }, nil } - return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil + return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-after-recheck"}, nil } respond := func(_ context.Context, answer extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) { responses++ @@ -66,7 +66,7 @@ func TestProjectWorkGateClarifiesBeforeProceeding(t *testing.T) { return core.HITLResponse{ChoiceID: "option-0"}, nil } decision, err := resolveProjectWorkDecision(context.Background(), "alice", "project-1", "run-1", "conversation.run", preflight, respond, ask) - if err != nil || decision.State != extension.ProjectWorkProceed || preflights != 1 || responses != 1 || questions != 1 { + if err != nil || decision.State != extension.ProjectWorkProceed || preflights != 2 || responses != 1 || questions != 1 { t.Fatalf("gate did not wait for clarification: decision=%+v preflight=%d responses=%d questions=%d err=%v", decision, preflights, responses, questions, err) } } @@ -99,9 +99,13 @@ func TestRunContextUsesHITLForProjectRiskBeforeContinuing(t *testing.T) { rc.h = &runHandle{hitlCh: make(chan hitlAnswer, 1)} rc.h.hitlCh <- hitlAnswer{resp: core.HITLResponse{ChoiceID: "option-0"}} rc.mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil }) - responded := false + responded, preflights := false, 0 rc.mgr.SetProjectWorkGate( func(context.Context, string, string) (extension.ProjectWorkDecision, error) { + preflights++ + if preflights > 1 { + return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-after-recheck"}, nil + } return extension.ProjectWorkDecision{State: extension.ProjectWorkClarify, RiskID: "risk-1", RiskRevision: "r1", Question: "效率具体指什么?", Options: []string{"效率更高"}}, nil }, func(_ context.Context, answer extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) { @@ -109,8 +113,8 @@ func TestRunContextUsesHITLForProjectRiskBeforeContinuing(t *testing.T) { return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-1"}, nil }, ) - if err := rc.requireProjectWorkDecision("conversation.run"); err != nil || !responded { - t.Fatalf("run did not wait for committed human clarification: responded=%t err=%v", responded, err) + if err := rc.requireProjectWorkDecision("conversation.run"); err != nil || !responded || preflights != 2 { + t.Fatalf("run did not recheck after committed human clarification: responded=%t preflights=%d err=%v", responded, preflights, err) } cards, err := st.ActiveHITLCards("alice", rc.run.ConversationID) if err != nil || len(cards) != 1 || cards[0].Question != "效率具体指什么?" {