revert: remove cross-agent worker direction synchronization
This commit is contained in:
1 parent
7e633b2056
commit
76210b9933
9 files changed
+11
-425
No files matched your search
@@ -339,7 +339,6 @@ const (
|
||||
IntentInterventionPending = "pending"
|
||||
IntentInterventionAccepted = "accepted"
|
||||
IntentInterventionRejected = "rejected"
|
||||
PlannerEventWorkerMessage = "worker_message"
|
||||
)
|
||||
|
||||
// IntentIntervention is the durable control record for one human interruption.
|
||||
@@ -356,19 +355,6 @@ type IntentIntervention struct {
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// WorkerDirection is the latest accepted human direction for one Worker
|
||||
// intent. The accepted activity is the durable planning event, so this view
|
||||
// survives restarts and is shared by Planner and Main Agent context builders.
|
||||
type WorkerDirection struct {
|
||||
ActivityID int64
|
||||
IntentID int64
|
||||
RequestID string
|
||||
Message string
|
||||
IntentState string
|
||||
IntentSummary string
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// lockIntentIntervention serializes request-id reservation/finalization across
|
||||
// processes. exploration_nodes.id is a global primary key, so the intent id is
|
||||
// sufficient input to the namespaced advisory key.
|
||||
@@ -577,7 +563,6 @@ WHERE a.exploration_id=$1 AND a.node_id=$2
|
||||
"intervention_request_id": requestID,
|
||||
"intervention_message": message,
|
||||
"intervention_status": IntentInterventionAccepted,
|
||||
"planner_event": PlannerEventWorkerMessage,
|
||||
"source": "user",
|
||||
})
|
||||
if err != nil {
|
||||
@@ -624,52 +609,6 @@ WHERE a.exploration_id=$1 AND a.node_id=$2
|
||||
return activity, false, nil
|
||||
}
|
||||
|
||||
// LatestWorkerDirections returns at most one accepted human direction per
|
||||
// local intent, newest directions first. It reads the visible user activity
|
||||
// rather than the hidden hand-off row so archive/restore and activity history
|
||||
// share one authoritative record.
|
||||
func (s *ExplorationStore) LatestWorkerDirections(limit int) ([]WorkerDirection, error) {
|
||||
if limit <= 0 {
|
||||
limit = 300
|
||||
}
|
||||
rows, err := s.db.Query(`
|
||||
SELECT activity_id,intent_id,request_id,message,intent_state,intent_summary,created_at
|
||||
FROM (
|
||||
SELECT DISTINCT ON (a.node_id)
|
||||
a.id AS activity_id,
|
||||
a.node_id AS intent_id,
|
||||
COALESCE(a.metadata->>'intervention_request_id','') AS request_id,
|
||||
COALESCE(a.metadata->>'intervention_message',a.detail,'') AS message,
|
||||
n.state AS intent_state,
|
||||
COALESCE(n.payload->>'summary','') AS intent_summary,
|
||||
a.created_at
|
||||
FROM activity a
|
||||
JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
|
||||
WHERE a.exploration_id=$1
|
||||
AND a.kind='user'
|
||||
AND n.kind='intent'
|
||||
AND a.metadata->>'intervention_status'=$2
|
||||
AND a.metadata->>'planner_event'=$3
|
||||
ORDER BY a.node_id,a.id DESC
|
||||
) latest
|
||||
ORDER BY activity_id DESC
|
||||
LIMIT $4`, s.expID, IntentInterventionAccepted, PlannerEventWorkerMessage, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make([]WorkerDirection, 0)
|
||||
for rows.Next() {
|
||||
var item WorkerDirection
|
||||
if err := rows.Scan(&item.ActivityID, &item.IntentID, &item.RequestID, &item.Message,
|
||||
&item.IntentState, &item.IntentSummary, &item.CreatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, item)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// RecoverableIntentInterventions returns control operations that must be
|
||||
// reconstructed before ordinary frontier claiming after a process restart:
|
||||
// pending transcript commits, plus accepted hand-offs not yet claimed once.
|
||||
|
||||
@@ -202,72 +202,3 @@ func TestIntentInterventionAcceptanceCanRetryAfterStateSettlement(t *testing.T)
|
||||
t.Fatalf("retry acceptance: activity=%+v duplicate=%v err=%v", activity, duplicate, err)
|
||||
}
|
||||
}
|
||||
|
||||
func acceptTestWorkerDirection(t *testing.T, store *ExplorationStore, intentID int64, requestID, message string) Activity {
|
||||
t.Helper()
|
||||
if claimed, err := store.ClaimIntent(intentID, "direction-test-worker"); err != nil || !claimed {
|
||||
t.Fatalf("claim direction intent: claimed=%v err=%v", claimed, err)
|
||||
}
|
||||
if _, err := store.ReserveIntentIntervention(intentID, requestID, message); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if changed, err := store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed {
|
||||
t.Fatalf("pause direction intent: changed=%v err=%v", changed, err)
|
||||
}
|
||||
activity, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message)
|
||||
if err != nil || duplicate {
|
||||
t.Fatalf("accept direction: activity=%+v duplicate=%v err=%v", activity, duplicate, err)
|
||||
}
|
||||
return activity
|
||||
}
|
||||
|
||||
func TestLatestWorkerDirectionsKeepsLatestPerIntentAcrossStoreReopen(t *testing.T) {
|
||||
d, err := Open(testDSN(t))
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) - skipping", err)
|
||||
}
|
||||
defer d.Close()
|
||||
task, err := d.CreateTask("worker direction persistence", "keep latest directions", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.DeleteTask(task.ID) //nolint:errcheck
|
||||
store := d.Exploration(task.ExplorationID)
|
||||
firstIntent, err := store.AddIntent(map[string]any{"summary": "first route"}, 1, nil, "planner")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
secondIntent, err := store.AddIntent(map[string]any{"summary": "second route"}, 1, nil, "planner")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
acceptTestWorkerDirection(t, store, firstIntent, "direction-old", "先检查旧方向")
|
||||
secondActivity := acceptTestWorkerDirection(t, store, secondIntent, "direction-second", "保留第二条意图方向")
|
||||
latestActivity := acceptTestWorkerDirection(t, store, firstIntent, "direction-latest", "改为验证最新权限边界")
|
||||
|
||||
d2, err := Open(testDSN(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d2.Close()
|
||||
directions, err := d2.Exploration(task.ExplorationID).LatestWorkerDirections(10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(directions) != 2 {
|
||||
t.Fatalf("directions=%+v, want one latest row per intent", directions)
|
||||
}
|
||||
byIntent := make(map[int64]WorkerDirection, len(directions))
|
||||
for _, direction := range directions {
|
||||
byIntent[direction.IntentID] = direction
|
||||
}
|
||||
if got := byIntent[firstIntent]; got.ActivityID != latestActivity.ID || got.RequestID != "direction-latest" ||
|
||||
got.Message != "改为验证最新权限边界" || got.IntentSummary != "first route" {
|
||||
t.Fatalf("latest first direction=%+v", got)
|
||||
}
|
||||
if got := byIntent[secondIntent]; got.ActivityID != secondActivity.ID || got.Message != "保留第二条意图方向" ||
|
||||
got.IntentSummary != "second route" {
|
||||
t.Fatalf("second direction=%+v", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user