weichen/Agent-System-Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Recheck project sources after clarification

Browse Source 
Why: Accepted retrieval and tool sources could become stale, and a pending HITL answer could release a changed source set.

Evidence: App and run-manager suites cover per-preflight resource verification, authorization epochs, and a second exact-source preflight after HITL.

Impact: Project work fails closed when authorization, source bytes, run binding, or the source watermark changes.

Refs: TT-N03 TT-R01
This commit is contained in:
weichen
2026-10-15 06:22:20 +00:00
parent c5c6416135
commit 99f1d3b3dd
4 changed files with 42 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/project_work_observation.go
+12 -1
View File
@@ -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,
internal/app/project_work_observation_test.go
+9 -5
View File
@@ -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)
}
}
internal/runmgr/project_work_gate.go
+12 -2
View File
@@ -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
}
internal/runmgr/project_work_gate_test.go
+9 -5
View File
@@ -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 != "效率具体指什么？" {