From 29b6eccb1d8b9abcf65ee4e38e0fac053f1c1d0f Mon Sep 17 00:00:00 2001 From: Grivn Date: Tue, 30 Jun 2026 12:39:03 +0800 Subject: [PATCH] fix: harden multica overlap validation Tighten Multica hub projection for complex parallel PoC runs by adding ledger reservations, ingest-sequence scoped feedback filtering, assignment-mailbox projection during managed wake, and richer acceptance checks for initial/follow-up child mailboxes. Validation: go test ./harness/cmd/mnemon-acceptance ./harness/cmd/mnemon-multica-runtime ./harness/internal/surface/multica ./harness/internal/driver; live parallel-poc-overlap acceptance passed on TEA-187 with five agents and four child mailboxes. --- .../acceptance_multica_runtime.go | 62 ++++++- .../acceptance_multica_runtime_test.go | 156 +++++++++++++++++- .../acceptance_multica_task_cases.go | 19 ++- .../acceptance_multica_task_cases_test.go | 5 + .../cmd/mnemon-multica-runtime/hub_writer.go | 147 ++++++++++++----- harness/cmd/mnemon-multica-runtime/main.go | 11 +- .../cmd/mnemon-multica-runtime/main_test.go | 92 ++++++++++- harness/internal/driver/multica_test.go | 64 +++++++ harness/internal/surface/multica/hub.go | 104 +++++++++--- .../internal/surface/multica/projection.go | 1 + 10 files changed, 579 insertions(+), 82 deletions(-) diff --git a/harness/cmd/mnemon-acceptance/acceptance_multica_runtime.go b/harness/cmd/mnemon-acceptance/acceptance_multica_runtime.go index 6bbb9fb6..28340d52 100644 --- a/harness/cmd/mnemon-acceptance/acceptance_multica_runtime.go +++ b/harness/cmd/mnemon-acceptance/acceptance_multica_runtime.go @@ -323,7 +323,7 @@ func runMulticaRuntimeProdSimAcceptance(ctx context.Context, opts multicaRuntime addMulticaProdSimAssertion(&report, "runtime recorded Mnemon ingest", strings.Contains(combined, "Mnemon ingest: recorded"), combined) } if opts.RequireManagedWake { - addMulticaProdSimAssertion(&report, "runtime completed managed wake", strings.Contains(combined, "Managed wake: completed"), combined) + addMulticaProdSimAssertion(&report, "runtime completed managed wake", multicaMessagesContainManagedWakeCompleted(combined), combined) } } if opts.RequireHubFlow { @@ -375,21 +375,42 @@ func multicaHubFlowManagedRuntimeReady(ctx context.Context, cli driver.MulticaCL if err != nil { return false, strings.Join(details, "; "), fmt.Errorf("read Multica agent env for %s: %w", principal, err) } - runtimeName := strings.TrimSpace(env["MNEMON_MANAGED_RUNTIME"]) - ready := multicaManagedRuntimeCanDriveTeamwork(runtimeName) + ready, label := multicaHubFlowParticipantReady(env) if ready { active++ } if principal == requiredPrincipal { requiredActive = ready } - details = append(details, fmt.Sprintf("%s=%s", principal, multicaManagedRuntimeReadinessLabel(runtimeName, ready))) + details = append(details, fmt.Sprintf("%s=%s", principal, label)) } ok := active >= minActive && requiredActive details = append(details, fmt.Sprintf("active=%d min=%d root_ready=%v", active, minActive, requiredActive)) return ok, strings.Join(details, "; "), nil } +func multicaHubFlowParticipantReady(env map[string]string) (bool, string) { + runtimeName := strings.TrimSpace(env["MNEMON_MANAGED_RUNTIME"]) + runtimeReady := multicaManagedRuntimeCanDriveTeamwork(runtimeName) + if !runtimeReady { + return false, multicaManagedRuntimeReadinessLabel(runtimeName, false) + } + var missing []string + if strings.TrimSpace(env["MNEMON_CONTROL_ADDR"]) == "" { + missing = append(missing, "MNEMON_CONTROL_ADDR") + } + if tokenFile := strings.TrimSpace(env["MNEMON_CONTROL_TOKEN_FILE"]); tokenFile != "" { + if info, err := os.Stat(tokenFile); err != nil || info.IsDir() { + missing = append(missing, "MNEMON_CONTROL_TOKEN_FILE") + } + } + label := multicaManagedRuntimeReadinessLabel(runtimeName, true) + if len(missing) > 0 { + return false, label + " (missing " + strings.Join(missing, ", ") + ")" + } + return true, label +} + func multicaManagedRuntimeCanDriveTeamwork(runtimeName string) bool { switch strings.ToLower(strings.TrimSpace(runtimeName)) { case "codex-appserver": @@ -424,7 +445,9 @@ func collectMulticaHubFlowEvidence(ctx context.Context, cli driver.MulticaCLI, o if minAssignmentChildren < 1 { minAssignmentChildren = 1 } - if report.TaskExpectations.MinChildMailboxes > minAssignmentChildren { + if report.TaskExpectations.InitialChildMailboxes > 0 { + minAssignmentChildren = report.TaskExpectations.InitialChildMailboxes + } else if report.TaskExpectations.MinChildMailboxes > minAssignmentChildren { minAssignmentChildren = report.TaskExpectations.MinChildMailboxes } children, childMeta, err := waitMulticaAssignmentChildren(ctx, cli, report.Issue.ID, opts.Wait, opts.Poll, minAssignmentChildren) @@ -450,6 +473,13 @@ func collectMulticaHubFlowEvidence(ctx context.Context, cli driver.MulticaCLI, o report.ChildMessages = childMessages report.ChildMessageTypes = multicaChildRunMessageTypeCounts(childMessages) report.ActiveAgents = activeAgents + if refreshedChildren, refreshedMeta, refreshErr := listMulticaAssignmentChildren(ctx, cli, report.Issue.ID); refreshErr == nil && len(refreshedChildren) > len(children) { + children = refreshedChildren + report.ChildIssues = refreshedChildren + for issueID, meta := range refreshedMeta { + report.ChildMetadata[issueID] = meta + } + } if err != nil { addMulticaProdSimAssertion(report, "hub-flow activates multiple Multica agents", false, err.Error()) collectMulticaHubFlowPartialSnapshot(ctx, cli, opts, report, children, err) @@ -462,7 +492,7 @@ func collectMulticaHubFlowEvidence(ctx context.Context, cli driver.MulticaCLI, o if strings.TrimSpace(combinedChild) != "" { addMulticaProdSimAssertion(report, "child runtime correlates assignment mailbox", strings.Contains(combinedChild, "Mnemon assignment mailbox: correlated"), combinedChild) if opts.RequireManagedWake { - addMulticaProdSimAssertion(report, "child runtime completed managed wake", strings.Contains(combinedChild, "Managed wake: completed"), combinedChild) + addMulticaProdSimAssertion(report, "child runtime completed managed wake", multicaMessagesContainManagedWakeCompleted(combinedChild), combinedChild) } } else { addMulticaProdSimAssertion(report, "child runtime correlates assignment mailbox", len(activeAgents) >= opts.MinActiveAgents, fmt.Sprintf("active_agents=%v messages deferred by Multica run state", activeAgents)) @@ -633,13 +663,21 @@ func listMulticaChildrenWithMetadata(ctx context.Context, cli driver.MulticaCLI, func waitMulticaChildRunEvidence(ctx context.Context, cli driver.MulticaCLI, rootIssueID string, rootRuns []driver.MulticaIssueRun, children []driver.MulticaIssue, wait, poll time.Duration, minActive int) (map[string][]driver.MulticaIssueRun, map[string][]driver.MulticaRunMessage, []string, error) { deadline := time.Now().Add(wait) + currentChildren := append([]driver.MulticaIssue(nil), children...) var lastRuns map[string][]driver.MulticaIssueRun var lastMessages map[string][]driver.MulticaRunMessage for { + refreshedChildren, _, err := listMulticaAssignmentChildren(ctx, cli, rootIssueID) + if err != nil { + return lastRuns, lastMessages, nil, err + } + if len(refreshedChildren) > len(currentChildren) { + currentChildren = refreshedChildren + } childRuns := map[string][]driver.MulticaIssueRun{} childMessages := map[string][]driver.MulticaRunMessage{} active := multicaActiveAgentIDs(rootRuns) - for _, child := range children { + for _, child := range currentChildren { runs, err := cli.ListIssueRuns(ctx, child.ID) if err != nil { return childRuns, childMessages, sortedMulticaActiveAgents(active), err @@ -664,11 +702,11 @@ func waitMulticaChildRunEvidence(ctx context.Context, cli driver.MulticaCLI, roo lastRuns = childRuns lastMessages = childMessages activeList := sortedMulticaActiveAgents(active) - if len(activeList) >= minActive && multicaEveryChildHasRun(childRuns, children) { + if len(activeList) >= minActive && multicaEveryChildHasRun(childRuns, currentChildren) { return childRuns, childMessages, activeList, nil } if wait <= 0 || time.Now().After(deadline) { - return lastRuns, lastMessages, activeList, fmt.Errorf("timed out waiting for child run evidence on root %s active_agents=%v child_messages=%d", rootIssueID, activeList, len(childMessages)) + return lastRuns, lastMessages, activeList, fmt.Errorf("timed out waiting for child run evidence on root %s children=%d active_agents=%v child_messages=%d", rootIssueID, len(currentChildren), activeList, len(childMessages)) } select { case <-ctx.Done(): @@ -724,6 +762,12 @@ func combinedMulticaChildMessages(messages map[string][]driver.MulticaRunMessage return strings.Join(parts, "\n") } +func multicaMessagesContainManagedWakeCompleted(text string) bool { + lower := strings.ToLower(text) + return strings.Contains(lower, "managed wake: completed") || + strings.Contains(lower, "managed wake completed") +} + func multicaRunMessageTypeCounts(messages []driver.MulticaRunMessage) map[string]int { out := map[string]int{} for _, message := range messages { diff --git a/harness/cmd/mnemon-acceptance/acceptance_multica_runtime_test.go b/harness/cmd/mnemon-acceptance/acceptance_multica_runtime_test.go index 444fe3a0..314a90c2 100644 --- a/harness/cmd/mnemon-acceptance/acceptance_multica_runtime_test.go +++ b/harness/cmd/mnemon-acceptance/acceptance_multica_runtime_test.go @@ -147,7 +147,7 @@ func TestMulticaRuntimeProdSimAcceptanceRequiresHubFlow(t *testing.T) { printf '%s\n' "$*" >> "$MULTICA_ARGS_PATH" cat >> "$MULTICA_STDIN_PATH" case "$*" in - *"agent env get agent-"*) agent="${4:-}"; printf '{"agent_id":"%s","custom_env":{"MNEMON_MANAGED_RUNTIME":"codex-appserver"}}\n' "$agent" ;; + *"agent env get agent-"*) agent="${4:-}"; printf '{"agent_id":"%s","custom_env":{"MNEMON_MANAGED_RUNTIME":"codex-appserver","MNEMON_CONTROL_ADDR":"http://127.0.0.1:8787"}}\n' "$agent" ;; *"issue create"*) printf '{"id":"root-9","identifier":"TEA-9","title":"Runtime hub flow","description":"Teamwork acceptance","status":"todo"}\n' ;; *"issue get root-9"*) printf '{"id":"root-9","identifier":"TEA-9","title":"Runtime hub flow","description":"Teamwork acceptance","status":"done"}\n' ;; *"issue get child-2"*) printf '%s\n' '{"id":"child-2","identifier":"TEA-10","title":"TEA-9: routing check","description":"## Assignment\n\nCheck routing.\n\n## Context\n\n- Root issue: [TEA-9](mention://issue/root-9) - Runtime hub flow\n- Assignee: researcher@team (mnemon-researcher)\n- Scope: routing check\n\n## Feedback\n\n- Expected feedback: result or blocker\n- Progress path: Mnemon runtime progress, result, or blocker feedback","status":"done"}' ;; @@ -228,6 +228,61 @@ esac } } +func TestWaitMulticaChildRunEvidenceRefreshesFollowupChildren(t *testing.T) { + tmp := t.TempDir() + argsPath := filepath.Join(tmp, "args.txt") + bin := filepath.Join(tmp, "multica") + script := `#!/usr/bin/env sh +printf '%s\n' "$*" >> "$MULTICA_ARGS_PATH" +case "$*" in + *"issue children root-1"*) printf '{"children":[{"id":"child-1","identifier":"TEA-2","title":"TEA-1: runtime","status":"done","metadata":{"mnemon.hub_backend":"multica","mnemon.kind":"assignment_mailbox","mnemon.root_issue_id":"root-1","mnemon.session_id":"multica:session:root-1","mnemon.assignment_id":"asg-1","mnemon.principal":"researcher@team"}},{"id":"child-2","identifier":"TEA-3","title":"TEA-1: runbook","status":"done","metadata":{"mnemon.hub_backend":"multica","mnemon.kind":"assignment_mailbox","mnemon.root_issue_id":"root-1","mnemon.session_id":"multica:session:root-1","mnemon.assignment_id":"asg-2","mnemon.principal":"implementer@team"}},{"id":"child-3","identifier":"TEA-4","title":"TEA-1: release","status":"done","metadata":{"mnemon.hub_backend":"multica","mnemon.kind":"assignment_mailbox","mnemon.root_issue_id":"root-1","mnemon.session_id":"multica:session:root-1","mnemon.assignment_id":"asg-3","mnemon.principal":"reviewer@team"}},{"id":"child-4","identifier":"TEA-5","title":"TEA-1: followup","status":"done","metadata":{"mnemon.hub_backend":"multica","mnemon.kind":"assignment_mailbox","mnemon.root_issue_id":"root-1","mnemon.session_id":"multica:session:root-1","mnemon.assignment_id":"asg-4","mnemon.principal":"integrator@team"}}]}\n' ;; + *"issue metadata list child-1"*) printf '[{"key":"mnemon.hub_backend","value":"multica"},{"key":"mnemon.kind","value":"assignment_mailbox"},{"key":"mnemon.root_issue_id","value":"root-1"},{"key":"mnemon.session_id","value":"multica:session:root-1"},{"key":"mnemon.assignment_id","value":"asg-1"},{"key":"mnemon.principal","value":"researcher@team"}]\n' ;; + *"issue metadata list child-2"*) printf '[{"key":"mnemon.hub_backend","value":"multica"},{"key":"mnemon.kind","value":"assignment_mailbox"},{"key":"mnemon.root_issue_id","value":"root-1"},{"key":"mnemon.session_id","value":"multica:session:root-1"},{"key":"mnemon.assignment_id","value":"asg-2"},{"key":"mnemon.principal","value":"implementer@team"}]\n' ;; + *"issue metadata list child-3"*) printf '[{"key":"mnemon.hub_backend","value":"multica"},{"key":"mnemon.kind","value":"assignment_mailbox"},{"key":"mnemon.root_issue_id","value":"root-1"},{"key":"mnemon.session_id","value":"multica:session:root-1"},{"key":"mnemon.assignment_id","value":"asg-3"},{"key":"mnemon.principal","value":"reviewer@team"}]\n' ;; + *"issue metadata list child-4"*) printf '[{"key":"mnemon.hub_backend","value":"multica"},{"key":"mnemon.kind","value":"assignment_mailbox"},{"key":"mnemon.root_issue_id","value":"root-1"},{"key":"mnemon.session_id","value":"multica:session:root-1"},{"key":"mnemon.assignment_id","value":"asg-4"},{"key":"mnemon.principal","value":"integrator@team"}]\n' ;; + *"issue runs child-1"*) printf '[{"id":"task-child-1","issue_id":"child-1","agent_id":"agent-researcher","status":"completed","completed_at":"2026-06-30T09:01:00Z"}]\n' ;; + *"issue runs child-2"*) printf '[{"id":"task-child-2","issue_id":"child-2","agent_id":"agent-implementer","status":"completed","completed_at":"2026-06-30T09:02:00Z"}]\n' ;; + *"issue runs child-3"*) printf '[{"id":"task-child-3","issue_id":"child-3","agent_id":"agent-reviewer","status":"completed","completed_at":"2026-06-30T09:03:00Z"}]\n' ;; + *"issue runs child-4"*) printf '[{"id":"task-child-4","issue_id":"child-4","agent_id":"agent-integrator","status":"completed","completed_at":"2026-06-30T09:04:00Z"}]\n' ;; + *"issue run-messages task-child-"*) printf '[{"seq":1,"type":"text","content":"Mnemon assignment mailbox: correlated. Managed wake: completed.","created_at":"2026-06-30T09:05:00Z"}]\n' ;; + *) printf '{}\n' ;; +esac +` + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("MULTICA_ARGS_PATH", argsPath) + + initialChildren := []driver.MulticaIssue{ + {ID: "child-1"}, + {ID: "child-2"}, + {ID: "child-3"}, + } + childRuns, _, activeAgents, err := waitMulticaChildRunEvidence( + context.Background(), + driver.MulticaCLI{Command: bin, WorkspaceID: "ws-1"}, + "root-1", + []driver.MulticaIssueRun{{ID: "task-root", IssueID: "root-1", AgentID: "agent-planner", Status: "completed", CompletedAt: "2026-06-30T09:00:00Z"}}, + initialChildren, + time.Second, + time.Millisecond, + 5, + ) + if err != nil { + t.Fatalf("wait child run evidence: %v", err) + } + if len(activeAgents) != 5 || len(childRuns["child-4"]) != 1 { + t.Fatalf("expected refreshed follow-up child evidence, active=%v runs=%+v", activeAgents, childRuns) + } + args, err := os.ReadFile(argsPath) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(args), "issue children root-1 --output json") || !strings.Contains(string(args), "issue runs child-4 --output json") { + t.Fatalf("expected child refresh and follow-up run lookup:\n%s", args) + } +} + func TestMulticaRuntimeProdSimHubFlowRejectsNoopManagedRuntimeBeforeCreatingIssue(t *testing.T) { tmp := t.TempDir() registryPath := filepath.Join(tmp, "registry.json") @@ -308,7 +363,7 @@ esac } } -func TestMulticaRuntimeProdSimHubFlowWritesPartialSnapshotWhenMailboxExpectationMisses(t *testing.T) { +func TestMulticaRuntimeProdSimHubFlowRejectsMissingControlAddrBeforeCreatingIssue(t *testing.T) { tmp := t.TempDir() registryPath := filepath.Join(tmp, "registry.json") var participants []driver.MulticaParticipantRecord @@ -337,6 +392,86 @@ printf '%s\n' "$*" >> "$MULTICA_ARGS_PATH" cat >> "$MULTICA_STDIN_PATH" case "$*" in *"agent env get agent-"*) agent="${4:-}"; printf '{"agent_id":"%s","custom_env":{"MNEMON_MANAGED_RUNTIME":"codex-appserver"}}\n' "$agent" ;; + *"issue create"*) printf 'issue create should not be reached\n' >&2; exit 99 ;; + *) printf '{}\n' ;; +esac +` + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("MULTICA_ARGS_PATH", argsPath) + t.Setenv("MULTICA_STDIN_PATH", stdinPath) + + report, err := runMulticaRuntimeProdSimAcceptance(context.Background(), multicaRuntimeProdSimOptions{ + RunRoot: filepath.Join(tmp, ".testdata", "multica-hub-no-control"), + MulticaBin: bin, + WorkspaceID: "ws-1", + RegistryPath: registryPath, + AssigneePrincipal: "planner@team", + IssueTitle: "No control hub flow", + IssueDescription: "Teamwork acceptance", + Wait: time.Millisecond, + Poll: time.Millisecond, + RequireIngest: true, + RequireHubFlow: true, + MinParticipants: 5, + MinActiveAgents: 3, + }) + if err == nil || !strings.Contains(err.Error(), "managed runtime participants") { + t.Fatalf("expected control prerequisite error, got err=%v report=%+v", err, report) + } + if report.Status != "failed" || report.Issue.ID != "" { + t.Fatalf("hub-flow prerequisite should fail before issue create: %+v", report) + } + var readinessAssertion *multicaRuntimeProdSimAssertion + for i := range report.Assertions { + if report.Assertions[i].Name == "hub-flow agents expose managed runtime" { + readinessAssertion = &report.Assertions[i] + break + } + } + if readinessAssertion == nil || readinessAssertion.Passed || !strings.Contains(readinessAssertion.Detail, "missing MNEMON_CONTROL_ADDR") { + t.Fatalf("unexpected readiness assertion: %+v all=%+v", readinessAssertion, report.Assertions) + } + rawArgs, err := os.ReadFile(argsPath) + if err != nil { + t.Fatal(err) + } + args := string(rawArgs) + if strings.Contains(args, "issue create") { + t.Fatalf("issue create must not run when hub-flow control addr is missing:\n%s", args) + } +} + +func TestMulticaRuntimeProdSimHubFlowWritesPartialSnapshotWhenMailboxExpectationMisses(t *testing.T) { + tmp := t.TempDir() + registryPath := filepath.Join(tmp, "registry.json") + var participants []driver.MulticaParticipantRecord + for _, role := range []string{"planner", "researcher", "implementer", "reviewer", "integrator"} { + participants = append(participants, driver.MulticaParticipantRecord{ + Principal: role + "@team", + AgentName: "mnemon-" + role, + AgentID: "agent-" + role, + Role: role, + }) + } + if err := driver.SaveMulticaRegistry(registryPath, driver.MulticaRegistry{ + SchemaVersion: 1, + WorkspaceID: "ws-1", + RuntimeProfileID: "profile-1", + RuntimeID: "runtime-1", + Participants: participants, + }); err != nil { + t.Fatal(err) + } + argsPath := filepath.Join(tmp, "args.txt") + stdinPath := filepath.Join(tmp, "stdin.txt") + bin := filepath.Join(tmp, "multica") + script := `#!/usr/bin/env sh +printf '%s\n' "$*" >> "$MULTICA_ARGS_PATH" +cat >> "$MULTICA_STDIN_PATH" +case "$*" in + *"agent env get agent-"*) agent="${4:-}"; printf '{"agent_id":"%s","custom_env":{"MNEMON_MANAGED_RUNTIME":"codex-appserver","MNEMON_CONTROL_ADDR":"http://127.0.0.1:8787"}}\n' "$agent" ;; *"issue create"*) printf '{"id":"root-partial","identifier":"TEA-30","title":"Partial hub evidence","description":"Teamwork acceptance","status":"todo"}\n' ;; *"issue get root-partial"*) printf '{"id":"root-partial","identifier":"TEA-30","title":"Partial hub evidence","description":"Teamwork acceptance","status":"in_progress"}\n' ;; *"issue get child-partial"*) printf '%s\n' '{"id":"child-partial","identifier":"TEA-31","title":"TEA-30: routing","description":"## Assignment\n\nCheck routing.\n\n## Context\n\n- Root issue: [TEA-30](mention://issue/root-partial) - Partial hub evidence\n- Assignee: researcher@team (mnemon-researcher)\n- Scope: routing\n\n## Feedback\n\n- Expected feedback: result","status":"done"}' ;; @@ -436,7 +571,7 @@ func TestMulticaRuntimeProdSimHubFlowAllowsDeferredRunMessages(t *testing.T) { printf '%s\n' "$*" >> "$MULTICA_ARGS_PATH" cat >> "$MULTICA_STDIN_PATH" case "$*" in - *"agent env get agent-"*) agent="${4:-}"; printf '{"agent_id":"%s","custom_env":{"MNEMON_MANAGED_RUNTIME":"codex-appserver"}}\n' "$agent" ;; + *"agent env get agent-"*) agent="${4:-}"; printf '{"agent_id":"%s","custom_env":{"MNEMON_MANAGED_RUNTIME":"codex-appserver","MNEMON_CONTROL_ADDR":"http://127.0.0.1:8787"}}\n' "$agent" ;; *"issue create"*) printf '{"id":"root-deferred","identifier":"TEA-20","title":"Deferred run messages","description":"Teamwork acceptance","status":"todo"}\n' ;; *"issue get root-deferred"*) printf '{"id":"root-deferred","identifier":"TEA-20","title":"Deferred run messages","description":"Teamwork acceptance","status":"done"}\n' ;; *"issue get child-a"*) printf '%s\n' '{"id":"child-a","identifier":"TEA-21","title":"TEA-20: routing","description":"## Assignment\n\nCheck routing.\n\n## Context\n\n- Root issue: [TEA-20](mention://issue/root-deferred) - Deferred run messages\n- Assignee: researcher@team (mnemon-researcher)\n- Scope: routing\n\n## Feedback\n\n- Expected feedback: result","status":"done"}' ;; @@ -485,6 +620,21 @@ esac } } +func TestMulticaMessagesContainManagedWakeCompleted(t *testing.T) { + cases := []string{ + "Managed wake: completed.", + "Managed wake completed: turn=turn-1.", + } + for _, text := range cases { + if !multicaMessagesContainManagedWakeCompleted(text) { + t.Fatalf("expected managed wake completion in %q", text) + } + } + if multicaMessagesContainManagedWakeCompleted("Managed wake failed: no candidate.") { + t.Fatal("failed wake should not be accepted as completed") + } +} + func TestMulticaRuntimeProdSimAcceptanceReportsPrerequisitesTogether(t *testing.T) { tmp := t.TempDir() report, err := runMulticaRuntimeProdSimAcceptance(context.Background(), multicaRuntimeProdSimOptions{ diff --git a/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases.go b/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases.go index da2a0e04..7711a941 100644 --- a/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases.go +++ b/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases.go @@ -33,10 +33,11 @@ type multicaAcceptanceTaskCaseMaterial struct { } type multicaAcceptanceTaskCaseExpectations struct { - MinActiveAgents int `json:"min_active_agents,omitempty"` - MinChildMailboxes int `json:"min_child_mailboxes,omitempty"` - MinFeedbackComments int `json:"min_feedback_comments,omitempty"` - TeamworkRounds []string `json:"teamwork_rounds,omitempty"` + MinActiveAgents int `json:"min_active_agents,omitempty"` + InitialChildMailboxes int `json:"initial_child_mailboxes,omitempty"` + MinChildMailboxes int `json:"min_child_mailboxes,omitempty"` + MinFeedbackComments int `json:"min_feedback_comments,omitempty"` + TeamworkRounds []string `json:"teamwork_rounds,omitempty"` } type multicaAcceptanceExecutionPlan struct { @@ -394,7 +395,7 @@ var multicaAcceptanceTaskCases = map[string]func(time.Time) multicaAcceptanceTas Title: "Parallel PoC overlap drill " + started.Format("150405"), Description: multicasurface.RootSessionDescription(multicasurface.RootSessionMaterial{ Request: "Run a production-like Mnemon-on-Multica collaboration case with three overlapping PoCs running in parallel. The goal is to validate Multica as the primary hub backend for assignment activation, feedback projection, and context reuse across related workstreams.", - WorkMode: "Use Multica root and child issues as the visible teamwork hub. The planner should split the case into parallel child assignment mailboxes, require shared context references in every feedback item, and create at least one follow-up assignment after reading first-round results.", + WorkMode: "Use Multica root and child issues as the visible teamwork hub, but create work only through Mnemon assignment events. Do not call Multica issue create, assign, rerun, or cancel commands to fan out work; the runtime projection creates child assignment mailboxes from accepted assignments. After recording Mnemon result or blocker feedback, workers should finish rather than waiting for their own Multica comment projection. The planner should split the case into parallel child assignment mailboxes, require shared context references in every feedback item, and create at least one follow-up assignment after reading first-round results.", Handoffs: []string{ "Round 1 - Observe: launch three parallel PoCs for runtime routing, operator runbook readiness, and release risk. Each PoC must name the shared context it consumed and the evidence it produced.", "Round 2 - Act: create a follow-up assignment for the highest disagreement or missing evidence across PoCs. The follow-up owner must reuse at least two shared contexts and cite prior child feedback.", @@ -410,9 +411,10 @@ var multicaAcceptanceTaskCases = map[string]func(time.Time) multicaAcceptanceTas Completion: "Finish only after the follow-up assignment feedback is visible, all three PoCs have result or blocker comments, and the root issue records an integrated decision that explains context reuse.", }), Expectations: multicaAcceptanceTaskCaseExpectations{ - MinActiveAgents: 5, - MinChildMailboxes: 4, - MinFeedbackComments: 4, + MinActiveAgents: 5, + InitialChildMailboxes: 3, + MinChildMailboxes: 4, + MinFeedbackComments: 4, TeamworkRounds: []string{ "Round 1 - Observe: three parallel PoCs split runtime, runbook, and release risk", "Round 2 - Act: follow up on disagreement or missing evidence across PoCs", @@ -526,6 +528,7 @@ var multicaAcceptanceTaskCases = map[string]func(time.Time) multicaAcceptanceTas "Every first-round feedback comment names at least one shared context and one evidence artifact.", "The follow-up assignment cites two prior child comments or artifacts before adding new work.", "The final root comment names which shared contexts were reused and where disagreement was resolved.", + "Direct Multica issue fan-out is invalid; all child work must originate from Mnemon assignment events and runtime mailbox projection.", "No visible issue text should expose session ids, assignment ids, assignment fingerprints, or projection-owner keys.", }, } diff --git a/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases_test.go b/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases_test.go index 6be534cc..b3062a0b 100644 --- a/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases_test.go +++ b/harness/cmd/mnemon-acceptance/acceptance_multica_task_cases_test.go @@ -143,6 +143,8 @@ esac } if report.TaskCase != multicaAcceptanceTaskCaseParallelPoc || report.TaskExpectations.MinActiveAgents != 5 || + report.TaskExpectations.InitialChildMailboxes != 3 || + report.TaskExpectations.MinChildMailboxes != 4 || len(report.ExecutionPlan.Workstreams) != 3 { t.Fatalf("task case report mismatch: %+v", report) } @@ -161,6 +163,9 @@ esac "## Execution Plan", "## Parallel PoCs", "## Context Reuse Checks", + "Do not call Multica issue create", + "finish rather than waiting for their own Multica comment projection", + "Direct Multica issue fan-out is invalid", "poc-runtime-routing", "poc-operator-runbook", "poc-release-risk", diff --git a/harness/cmd/mnemon-multica-runtime/hub_writer.go b/harness/cmd/mnemon-multica-runtime/hub_writer.go index 7713aeee..8accffed 100644 --- a/harness/cmd/mnemon-multica-runtime/hub_writer.go +++ b/harness/cmd/mnemon-multica-runtime/hub_writer.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sort" + "strconv" "strings" "time" @@ -168,6 +169,19 @@ func (s *runtimeRPCState) projectAssignmentMailboxes(ctx context.Context, cli dr var errs []error for _, projection := range projections { material := assignmentMailboxMaterial(projection.Item, projection.Result, projection.RootIssue, projection.Participant) + reservation := driver.MulticaHubLedgerRecord{ + Kind: driver.MulticaHubKindAssignmentMailbox, + Source: projection.Source, + Target: driver.MulticaHubLedgerTarget{ + RootIssueID: projection.Result.RootIssueID, + Status: "reserved", + }, + } + if _, reserved, err := ledger.Reserve(reservation); err != nil { + return created, err + } else if !reserved { + continue + } child, err := retryMulticaHubValue(ctx, func() (driver.MulticaIssue, error) { return cli.CreateIssue(ctx, driver.MulticaCreateIssueRequest{ Title: multicasurface.AssignmentMailboxTitle(material), @@ -182,18 +196,6 @@ func (s *runtimeRPCState) projectAssignmentMailboxes(ctx context.Context, cli dr errs = append(errs, fmt.Errorf("create assignment mailbox %s: %w", projection.Item.ID, err)) continue } - fullMeta := projection.Metadata.Map() - dispatchMeta := multicasurface.AssignmentMailboxDispatchMetadata(fullMeta) - if err := setMulticaHubMetadataMap(ctx, cli, child.ID, dispatchMeta); err != nil { - errs = append(errs, fmt.Errorf("tag assignment mailbox %s (%s): %w", projection.Item.ID, child.ID, err)) - continue - } - if _, err := retryMulticaHubValue(ctx, func() (driver.MulticaIssue, error) { - return cli.AssignIssue(ctx, child.ID, projection.Participant.AgentID) - }); err != nil { - errs = append(errs, fmt.Errorf("assign assignment mailbox %s (%s): %w", projection.Item.ID, child.ID, err)) - continue - } if err := ledger.Record(driver.MulticaHubLedgerRecord{ Kind: driver.MulticaHubKindAssignmentMailbox, Source: projection.Source, @@ -206,6 +208,18 @@ func (s *runtimeRPCState) projectAssignmentMailboxes(ctx context.Context, cli dr return created, err } created++ + fullMeta := projection.Metadata.Map() + dispatchMeta := multicasurface.AssignmentMailboxDispatchMetadata(fullMeta) + if err := setMulticaHubMetadataMap(ctx, cli, child.ID, dispatchMeta); err != nil { + errs = append(errs, fmt.Errorf("tag assignment mailbox %s (%s): %w", projection.Item.ID, child.ID, err)) + continue + } + if _, err := retryMulticaHubValue(ctx, func() (driver.MulticaIssue, error) { + return cli.AssignIssue(ctx, child.ID, projection.Participant.AgentID) + }); err != nil { + errs = append(errs, fmt.Errorf("assign assignment mailbox %s (%s): %w", projection.Item.ID, child.ID, err)) + continue + } if err := setMulticaHubMetadataMap(ctx, cli, child.ID, multicasurface.AssignmentMailboxSupplementalMetadata(fullMeta, dispatchMeta)); err != nil { errs = append(errs, fmt.Errorf("tag supplemental assignment mailbox %s (%s): %w", projection.Item.ID, child.ID, err)) continue @@ -271,56 +285,63 @@ func (s *runtimeRPCState) writeProgressComments(ctx context.Context, cli driver. if item.ID == "" || item.AssignmentRef == "" { continue } - if !runtimeItemAfterRootIngest(item.IngestSeq, result) { - continue - } progressProjection := multicasurface.ProgressFeedbackProjectionForRuntimeItem(multicasurface.ProgressFeedbackProjectionMaterial{ Item: item, SessionID: result.SessionID, CorrelationID: result.CorrelationID, }) - if rec, ok, err := ledger.Find(driver.MulticaHubKindFeedbackCarrier, progressProjection.Source); err != nil { + rec, recorded, err := ledger.Find(driver.MulticaHubKindFeedbackCarrier, progressProjection.Source) + if err != nil { return err - } else if ok { - child := strings.TrimSpace(rec.Target.ChildIssueID) - if child == "" { - var found bool - child, found, err = findAssignmentTargetFromLedger(ledger, result.SessionID, item.AssignmentRef, item.Actor) + } + child := strings.TrimSpace(rec.Target.ChildIssueID) + if child == "" { + var found bool + child, found, err = findAssignmentTargetFromLedger(ledger, result.SessionID, item.AssignmentRef, item.Actor) + if err != nil { + return err + } + if !found { + child, found, err = findAssignmentTargetFromMulticaHub(ctx, cli, result.RootIssueID, result.SessionID, item.AssignmentRef, item.Actor) if err != nil { return err } - if !found { - child, found, err = findAssignmentTargetFromMulticaHub(ctx, cli, result.RootIssueID, result.SessionID, item.AssignmentRef, item.Actor) - if err != nil { - return err - } - } - if !found { - continue - } } - if !runtimeProgressMatchesCurrentMulticaScope(item, result, child) { + if !found { continue } - if err := s.ensureProgressIssueStatuses(ctx, cli, result, child, progressProjection.Feedback); err != nil { - return err - } + } + if !runtimeProgressMatchesCurrentMulticaScope(item, result, child) { continue } - child, ok, err := findAssignmentTargetFromLedger(ledger, result.SessionID, item.AssignmentRef, item.Actor) + ok, err := runtimeProgressAfterAssignmentMailbox(ledger, result.SessionID, item.AssignmentRef, item.Actor, child, item.IngestSeq) if err != nil { return err } if !ok { - child, ok, err = findAssignmentTargetFromMulticaHub(ctx, cli, result.RootIssueID, result.SessionID, item.AssignmentRef, item.Actor) - if err != nil { + continue + } + if recorded { + if err := s.ensureProgressIssueStatuses(ctx, cli, result, child, progressProjection.Feedback); err != nil { return err } - } - if !ok { continue } - if !runtimeProgressMatchesCurrentMulticaScope(item, result, child) { + reservation := driver.MulticaHubLedgerRecord{ + Kind: driver.MulticaHubKindFeedbackCarrier, + Source: progressProjection.Source, + Target: driver.MulticaHubLedgerTarget{ + RootIssueID: result.RootIssueID, + ChildIssueID: child, + Status: "reserved", + }, + } + if _, reserved, err := ledger.Reserve(reservation); err != nil { + return err + } else if !reserved { + if err := s.ensureProgressIssueStatuses(ctx, cli, result, child, progressProjection.Feedback); err != nil { + return err + } continue } comment, err := cli.AddIssueComment(ctx, child, progressProjection.CommentBody) @@ -347,6 +368,52 @@ func (s *runtimeRPCState) writeProgressComments(ctx context.Context, cli driver. return nil } +func runtimeProgressAfterAssignmentMailbox(ledger *driver.FileMulticaHubLedger, sessionID, assignmentID, principal, childIssueID string, progressSeq int64) (bool, error) { + if progressSeq <= 0 { + return true, nil + } + records, err := ledger.Records() + if err != nil { + return false, err + } + sessionID = strings.TrimSpace(sessionID) + assignmentID = strings.TrimSpace(assignmentID) + principal = strings.TrimSpace(principal) + childIssueID = strings.TrimSpace(childIssueID) + for i := len(records) - 1; i >= 0; i-- { + record := records[i] + if record.Kind != driver.MulticaHubKindAssignmentMailbox || + strings.TrimSpace(record.Source.SessionID) != sessionID || + strings.TrimSpace(record.Source.AssignmentID) != assignmentID || + strings.TrimSpace(record.Source.Principal) != principal || + strings.TrimSpace(record.Target.ChildIssueID) != childIssueID { + continue + } + assignmentSeq := record.Source.IngestSeq + if assignmentSeq <= 0 { + assignmentSeq = multicaLocalEventSeq(record.Source.EventID) + } + if assignmentSeq <= 0 { + return true, nil + } + return progressSeq > assignmentSeq, nil + } + return true, nil +} + +func multicaLocalEventSeq(eventID string) int64 { + eventID = strings.TrimSpace(eventID) + if eventID == "" { + return 0 + } + parts := strings.Split(eventID, "/") + if len(parts) == 0 { + return 0 + } + seq, _ := strconv.ParseInt(parts[len(parts)-1], 10, 64) + return seq +} + func (s *runtimeRPCState) ensureProgressIssueStatuses(ctx context.Context, cli driver.MulticaCLI, result *runtimeImportResult, child string, material multicasurface.ProgressFeedbackMaterial) error { if status := multicasurface.ProgressIssueStatus(material); status != "" { if err := retryMulticaHubOperation(ctx, func() error { diff --git a/harness/cmd/mnemon-multica-runtime/main.go b/harness/cmd/mnemon-multica-runtime/main.go index 956d1da5..3bb8f078 100644 --- a/harness/cmd/mnemon-multica-runtime/main.go +++ b/harness/cmd/mnemon-multica-runtime/main.go @@ -654,7 +654,7 @@ func (d runtimeHubProjectionDelta) active() bool { func (s *runtimeRPCState) wakeManagedAgentWithHubProjection(ctx context.Context, cli driver.MulticaCLI, client *access.Client, rootIssue driver.MulticaIssue, result *runtimeImportResult, progress runtimeProgressSink) []runtimeHubProjectionDelta { if result == nil || client == nil || !strings.EqualFold(result.HubBackend, driver.MulticaHubBackend) || - result.HubKind != driver.MulticaHubKindSession || + !runtimeManagedWakeHubProjectionKind(result.HubKind) || !multicasurface.RuntimeHubWriteEnabled(s.Env) { s.wakeManagedAgent(result, progress) return nil @@ -689,6 +689,15 @@ func (s *runtimeRPCState) wakeManagedAgentWithHubProjection(ctx context.Context, return out } +func runtimeManagedWakeHubProjectionKind(kind string) bool { + switch strings.TrimSpace(kind) { + case driver.MulticaHubKindSession, driver.MulticaHubKindAssignmentMailbox: + return true + default: + return false + } +} + func (s *runtimeRPCState) writeMulticaHubArtifactsDelta(ctx context.Context, cli driver.MulticaCLI, client *access.Client, rootIssue driver.MulticaIssue, base runtimeImportResult) runtimeHubProjectionDelta { snapshot := base snapshot.HubChildIssues = 0 diff --git a/harness/cmd/mnemon-multica-runtime/main_test.go b/harness/cmd/mnemon-multica-runtime/main_test.go index 6894d681..dbc9fef8 100644 --- a/harness/cmd/mnemon-multica-runtime/main_test.go +++ b/harness/cmd/mnemon-multica-runtime/main_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "encoding/json" "net/http" "net/http/httptest" @@ -37,6 +38,81 @@ func TestMergeRuntimeHubProjectionDeltasPreservesEarlyDispatchCounts(t *testing. } } +func TestRuntimeItemAfterRootIngestFiltersHistoricalAssignments(t *testing.T) { + receipt := &access.IngestReceipt{Seq: 50} + if runtimeItemAfterRootIngest(49, &runtimeImportResult{Receipt: receipt, HubKind: driver.MulticaHubKindSession}) { + t.Fatal("session hub write should ignore assignments older than the root ingest receipt") + } + if !runtimeItemAfterRootIngest(51, &runtimeImportResult{Receipt: receipt, HubKind: driver.MulticaHubKindSession}) { + t.Fatal("session hub write should project assignments after the root ingest receipt") + } +} + +func TestRuntimeManagedWakeHubProjectionIncludesAssignmentMailboxes(t *testing.T) { + for _, kind := range []string{driver.MulticaHubKindSession, driver.MulticaHubKindAssignmentMailbox} { + if !runtimeManagedWakeHubProjectionKind(kind) { + t.Fatalf("hub projection should run during managed wake for %s", kind) + } + } + if runtimeManagedWakeHubProjectionKind(driver.MulticaHubKindFeedbackCarrier) { + t.Fatal("feedback carriers are projected artifacts, not managed wake issue kinds") + } +} + +func TestProjectAssignmentMailboxesSkipsExistingReservation(t *testing.T) { + ledger := driver.NewFileMulticaHubLedger(filepath.Join(t.TempDir(), "hub-ledger.jsonl")) + source := driver.MulticaHubLedgerSource{ + SessionID: "session-1", + CorrelationID: "corr-1", + EventID: "event-1", + AssignmentID: "assignment-1", + AssignmentFingerprint: "sha256:abc", + Principal: "worker@team", + ProjectionKind: "assignment", + } + if _, ok, err := ledger.Reserve(driver.MulticaHubLedgerRecord{ + Kind: driver.MulticaHubKindAssignmentMailbox, + Source: source, + Target: driver.MulticaHubLedgerTarget{ + RootIssueID: "root-1", + Status: "reserved", + }, + }); err != nil || !ok { + t.Fatalf("reserve: ok=%v err=%v", ok, err) + } + created, err := (&runtimeRPCState{}).projectAssignmentMailboxes(context.Background(), driver.MulticaCLI{ + Command: filepath.Join(t.TempDir(), "missing-multica"), + }, ledger, []runtimeAssignmentProjection{{ + Item: multicasurface.RuntimeAssignmentItem{ + ID: "assignment-1", + Assignee: "worker@team", + ExpectedWork: "check routing", + ExpectedFeedback: "progress digest", + }, + Participant: driver.MulticaParticipantRecord{ + Principal: "worker@team", + AgentID: "agent-1", + AgentName: "worker", + }, + Source: source, + Metadata: multicasurface.MulticaHubMetadata{ + Kind: driver.MulticaHubKindAssignmentMailbox, + }, + RootIssue: driver.MulticaIssue{ID: "root-1", Identifier: "TEA-1", Title: "Root"}, + Result: &runtimeImportResult{ + RootIssueID: "root-1", + SessionID: "session-1", + Principal: "planner@team", + }, + }}) + if err != nil { + t.Fatal(err) + } + if created != 0 { + t.Fatalf("created = %d, want 0", created) + } +} + func TestManagedWakeMatchTermsPreferStableIssueIdentity(t *testing.T) { got := drive.ManagedWakeMatchTerms(runtimeManagedWakeMatchMaterial(runtimeImportResult{ IssueID: "issue-123", @@ -685,7 +761,9 @@ esac t.Fatalf("runtime comment must not expose hub backend routing details:\n%s", comment) } args := mustReadRuntimeTestFile(t, argsPath) - if !strings.Contains(args, "issue metadata set iss-9 --key mnemon.hub_backend") || !strings.Contains(args, "issue comment add iss-9 --content-stdin --output json") { + metadataAttempt := strings.Index(args, "issue metadata set iss-9 --key mnemon.") + commentProjection := strings.Index(args, "issue comment add iss-9 --content-stdin --output json") + if metadataAttempt < 0 || commentProjection < 0 || metadataAttempt > commentProjection { t.Fatalf("runtime should attempt metadata projection then continue to comment projection:\n%s", args) } } @@ -799,6 +877,15 @@ esac }, { Ref: contract.ResourceRef{Kind: "progress_digest", ID: "project"}, Fields: map[string]any{"items": []any{map[string]any{ + "id": "pg-old-writer", + "event_id": "pg-old-writer", + "ingest_seq": float64(31), + "actor": "worker@team", + "assignment_ref": "asg-writer", + "feedback_kind": "progress", + "summary": "Old writer progress must not attach to the new mailbox.", + "artifact_refs": []any{"multica:issue:root-2"}, + }, map[string]any{ "id": "pg-writer", "event_id": "pg-writer", "ingest_seq": float64(33), @@ -959,6 +1046,9 @@ esac t.Fatalf("child comment must not expose machine field %q:\n%s", blocked, childComment) } } + if strings.Contains(childComment, "Old writer progress") { + t.Fatalf("old progress attached to new assignment mailbox:\n%s", childComment) + } } func TestRuntimeRoutesDuplicateAssignmentIDProgressByPrincipal(t *testing.T) { diff --git a/harness/internal/driver/multica_test.go b/harness/internal/driver/multica_test.go index 83d1fe2e..0cc51d61 100644 --- a/harness/internal/driver/multica_test.go +++ b/harness/internal/driver/multica_test.go @@ -5,6 +5,8 @@ import ( "os" "path/filepath" "strings" + "sync" + "sync/atomic" "testing" ) @@ -55,6 +57,21 @@ printf '{"id":"comment-1","issue_id":"iss-1","content":"ok","type":"comment"}\n' } } +func TestDecodeMulticaCommentListOutputDedupesNestedComments(t *testing.T) { + comments, err := decodeMulticaCommentListOutput([]byte(`{ + "comments": [ + {"id":"comment-1","issue_id":"issue-1","content":"ok","type":"comment"}, + {"id":"comment-1","issue_id":"issue-1","content":"ok","type":"comment"} + ] + }`)) + if err != nil { + t.Fatal(err) + } + if len(comments) != 1 || comments[0].ID != "comment-1" { + t.Fatalf("comments = %+v", comments) + } +} + func TestMulticaCLIAuthStatusFallsBackToStderr(t *testing.T) { tmp := t.TempDir() bin := filepath.Join(tmp, "multica") @@ -447,6 +464,53 @@ func TestFileMulticaHubLedgerDedupesRecords(t *testing.T) { } } +func TestFileMulticaHubLedgerReserveIsAtomicAcrossInstances(t *testing.T) { + path := filepath.Join(t.TempDir(), "hub-ledger.jsonl") + source := MulticaHubLedgerSource{ + SessionID: "session-1", + EventID: "progress-1", + AssignmentID: "assignment-1", + Principal: "worker@team", + ProjectionKind: "progress", + } + record := MulticaHubLedgerRecord{ + Kind: MulticaHubKindFeedbackCarrier, + Source: source, + Target: MulticaHubLedgerTarget{ + RootIssueID: "root-1", + ChildIssueID: "child-1", + Status: "reserved", + }, + } + var created int64 + var wg sync.WaitGroup + for i := 0; i < 16; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, ok, err := NewFileMulticaHubLedger(path).Reserve(record) + if err != nil { + t.Errorf("reserve: %v", err) + return + } + if ok { + atomic.AddInt64(&created, 1) + } + }() + } + wg.Wait() + if created != 1 { + t.Fatalf("reserve winners = %d, want 1", created) + } + records, err := NewFileMulticaHubLedger(path).Records() + if err != nil { + t.Fatal(err) + } + if len(records) != 1 || records[0].Target.Status != "reserved" { + t.Fatalf("reservation records = %+v", records) + } +} + func TestMulticaCLIChildrenAndMetadataCommands(t *testing.T) { tmp := t.TempDir() argsPath := filepath.Join(tmp, "args.txt") diff --git a/harness/internal/surface/multica/hub.go b/harness/internal/surface/multica/hub.go index b1dbd4fb..2579fc1d 100644 --- a/harness/internal/surface/multica/hub.go +++ b/harness/internal/surface/multica/hub.go @@ -11,6 +11,8 @@ import ( "sort" "strings" "time" + + "golang.org/x/sys/unix" ) const ( @@ -114,6 +116,7 @@ type MulticaHubLedgerSource struct { SessionID string `json:"session_id,omitempty"` CorrelationID string `json:"correlation_id,omitempty"` EventID string `json:"event_id,omitempty"` + IngestSeq int64 `json:"ingest_seq,omitempty"` AssignmentID string `json:"assignment_id,omitempty"` AssignmentFingerprint string `json:"assignment_fingerprint,omitempty"` Principal string `json:"principal,omitempty"` @@ -157,30 +160,61 @@ func NewFileMulticaHubLedger(path string) *FileMulticaHubLedger { } func (l *FileMulticaHubLedger) Records() ([]MulticaHubLedgerRecord, error) { - if err := l.load(); err != nil { - return nil, err - } - return append([]MulticaHubLedgerRecord(nil), l.records...), nil + var records []MulticaHubLedgerRecord + err := l.withLockedLoad(func() error { + records = append([]MulticaHubLedgerRecord(nil), l.records...) + return nil + }) + return records, err } func (l *FileMulticaHubLedger) Find(kind string, source MulticaHubLedgerSource) (MulticaHubLedgerRecord, bool, error) { - if err := l.load(); err != nil { - return MulticaHubLedgerRecord{}, false, err - } - want := multicaHubLedgerKey(kind, source) - for i := len(l.records) - 1; i >= 0; i-- { - rec := l.records[i] - if multicaHubLedgerKey(rec.Kind, rec.Source) == want { - return rec, true, nil + var found MulticaHubLedgerRecord + var ok bool + err := l.withLockedLoad(func() error { + want := multicaHubLedgerKey(kind, source) + for i := len(l.records) - 1; i >= 0; i-- { + rec := l.records[i] + if multicaHubLedgerKey(rec.Kind, rec.Source) == want { + found = rec + ok = true + return nil + } } - } - return MulticaHubLedgerRecord{}, false, nil + return nil + }) + return found, ok, err } func (l *FileMulticaHubLedger) Record(record MulticaHubLedgerRecord) error { - if err := l.load(); err != nil { - return err - } + return l.withLockedLoad(func() error { + return l.recordLoaded(record) + }) +} + +func (l *FileMulticaHubLedger) Reserve(record MulticaHubLedgerRecord) (MulticaHubLedgerRecord, bool, error) { + var reserved MulticaHubLedgerRecord + var created bool + err := l.withLockedLoad(func() error { + key := multicaHubLedgerKey(record.Kind, record.Source) + for i := len(l.records) - 1; i >= 0; i-- { + rec := l.records[i] + if multicaHubLedgerKey(rec.Kind, rec.Source) == key { + reserved = rec + return nil + } + } + if err := l.recordLoaded(record); err != nil { + return err + } + reserved = record + created = true + return nil + }) + return reserved, created, err +} + +func (l *FileMulticaHubLedger) recordLoaded(record MulticaHubLedgerRecord) error { if strings.TrimSpace(l.path) == "" { return fmt.Errorf("multica hub ledger path is required") } @@ -204,6 +238,14 @@ func (l *FileMulticaHubLedger) Record(record MulticaHubLedgerRecord) error { break } } + if err := l.appendRecord(record); err != nil { + return err + } + l.upsertLoaded(record) + return nil +} + +func (l *FileMulticaHubLedger) appendRecord(record MulticaHubLedgerRecord) error { if err := os.MkdirAll(filepath.Dir(l.path), 0o755); err != nil { return err } @@ -216,11 +258,32 @@ func (l *FileMulticaHubLedger) Record(record MulticaHubLedgerRecord) error { _ = f.Close() return err } - if err := f.Close(); err != nil { + return f.Close() +} + +func (l *FileMulticaHubLedger) withLockedLoad(fn func() error) error { + if strings.TrimSpace(l.path) == "" { + return fmt.Errorf("multica hub ledger path is required") + } + if err := os.MkdirAll(filepath.Dir(l.path), 0o755); err != nil { + return err + } + lock, err := os.OpenFile(l.path+".lock", os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { return err } - l.upsertLoaded(record) - return nil + defer lock.Close() + if err := unix.Flock(int(lock.Fd()), unix.LOCK_EX); err != nil { + return err + } + defer unix.Flock(int(lock.Fd()), unix.LOCK_UN) + l.loaded = false + l.loadErr = nil + l.records = nil + if err := l.load(); err != nil { + return err + } + return fn() } func (l *FileMulticaHubLedger) load() error { @@ -488,6 +551,7 @@ func AssignmentMailboxProjectionForRuntimeItem(material AssignmentMailboxProject SessionID: material.SessionID, CorrelationID: material.CorrelationID, EventID: item.EventID, + IngestSeq: item.IngestSeq, AssignmentID: item.ID, AssignmentFingerprint: fingerprint, Principal: item.Assignee, diff --git a/harness/internal/surface/multica/projection.go b/harness/internal/surface/multica/projection.go index b420c52e..179fb780 100644 --- a/harness/internal/surface/multica/projection.go +++ b/harness/internal/surface/multica/projection.go @@ -185,6 +185,7 @@ func ProgressFeedbackProjectionForRuntimeItem(material ProgressFeedbackProjectio SessionID: material.SessionID, CorrelationID: material.CorrelationID, EventID: item.EventID, + IngestSeq: item.IngestSeq, AssignmentID: item.AssignmentRef, Principal: item.Actor, ProjectionKind: "progress",