package agenttask import ( "context" "fmt" "strings" "testing" "time" ) func TestManagerS03S16MultiProjectManualResumeAndParallelTrace(t *testing.T) { projectA := testSnapshot( "project-a", "shared-workspace", testUnit("a-disjoint", WriteSetDisjoint), testUnit("a-overlap", WriteSetOverlap), ) projectB := testSnapshot( "project-b", "workspace-b", testUnit("b-unknown", WriteSetUnknown), ) unselected := testSnapshot( "unselected", "workspace-unselected", testUnit("must-not-run", WriteSetUnknown), ) harness := newHarness( t, map[ProjectID]ProjectWorkflowSnapshot{ "project-a": projectA, "project-b": projectB, "unselected": unselected, }, 3, ) for _, workID := range []WorkUnitID{"a-disjoint", "a-overlap", "b-unknown"} { harness.invoker.delays[workID] = 20 * time.Millisecond } harness.start("project-a", "shared-workspace", nil) harness.start("project-b", "workspace-b", nil) if err := harness.manager.Reconcile(context.Background()); err != nil { t.Fatalf("Reconcile: %v", err) } state := harness.store.snapshot() if state.Projects["project-a"].Status != ProjectStatusCompleted || state.Projects["project-b"].Status != ProjectStatusCompleted { t.Fatalf( "manual projects = %s/%s", state.Projects["project-a"].Status, state.Projects["project-b"].Status, ) } if state.Projects["unselected"].Status != ProjectStatusObserved { t.Fatalf("unselected project status = %s", state.Projects["unselected"].Status) } if harness.invoker.callCount() != 3 { t.Fatalf("invocations = %d, want selected work only", harness.invoker.callCount()) } if harness.invoker.maxConcurrency() < 2 { t.Fatalf("max concurrency = %d, want isolated parallel dispatch", harness.invoker.maxConcurrency()) } trace := make([]string, 0) for _, event := range harness.events.snapshot() { if event.Type != EventDispatchStarted && event.Type != EventIntegrationResult { continue } trace = append(trace, fmt.Sprintf( "%s:%s:%s:%d:%s:%s", event.Type, event.ProjectID, event.WorkUnitID, event.Ordinal, event.WriteSetKind, event.IsolationMode, )) if event.WorkUnitID == "must-not-run" { t.Fatalf("unselected work appeared in execution trace: %v", trace) } } joined := strings.Join(trace, "\n") for _, expected := range []string{"a-disjoint", "a-overlap", "b-unknown"} { if !strings.Contains(joined, expected) { t.Fatalf("trace missing %s:\n%s", expected, joined) } } t.Logf( "S03 trace: unselected_invocations=0 manual_projects=2 terminal=2; "+ "S16 trace: explicit_dependency_only=true isolated_parallel_max=%d integration_ordinals=%v\n%s", harness.invoker.maxConcurrency(), harness.integrator.ordinals(), joined, ) } func TestManagerWorkflowEvidenceGateAndPiRepair(t *testing.T) { t.Run("complete evidence invokes review", func(t *testing.T) { harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{ "project": testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown)), }, 1) harness.start("project", "workspace", nil) if err := harness.manager.Reconcile(context.Background()); err != nil { t.Fatalf("Reconcile: %v", err) } if harness.reviewer.callCount() != 1 { t.Fatalf("review calls = %d, want 1", harness.reviewer.callCount()) } }) for _, test := range []struct { name string configure func(*managerHarness) wantCode BlockerCode wantRepair int }{ { name: "placeholder from another provider never repairs or reviews", configure: func(harness *managerHarness) { harness.evidence.observeFunc = func(request WorkflowEvidenceRequest, _ int) (ArtifactEvidence, error) { return ArtifactEvidence{ Active: true, Identity: artifactIdentity(request.Project, request.Work, request.Submission), Completeness: ArtifactPlaceholder, }, nil } }, wantCode: BlockerEvidenceRepairDenied, }, { name: "wrong active artifact identity never invokes review", configure: func(harness *managerHarness) { harness.evidence.observeFunc = func(request WorkflowEvidenceRequest, _ int) (ArtifactEvidence, error) { identity := artifactIdentity(request.Project, request.Work, request.Submission) identity.AttemptID = "stale-attempt" return ArtifactEvidence{Active: true, Identity: identity, Completeness: ArtifactComplete}, nil } }, wantCode: BlockerArtifactMismatch, }, { name: "Pi repair rejects a stale native locator", configure: func(harness *managerHarness) { harness.manager.selector = fakeSelector{capacity: 1, providerID: "pi"} harness.invoker.locators["work"] = []LocatorRecord{{ Kind: LocatorSession, Opaque: "native-session", Revision: "session-r1", ProjectID: "project", WorkspaceID: "workspace", WorkUnitID: "work", AttemptID: attemptID("work", 1), }} harness.evidence.observeFunc = func(request WorkflowEvidenceRequest, _ int) (ArtifactEvidence, error) { identity := artifactIdentity(request.Project, request.Work, request.Submission) return ArtifactEvidence{ Active: true, Identity: identity, Completeness: ArtifactPlaceholder, RepairIntent: &EvidenceRepairIntent{ Identity: identity, DispatchOrdinal: request.Work.DispatchOrdinal, NativeLocator: LocatorRecord{ Kind: LocatorSession, Opaque: "stale-session", Revision: "session-r1", ProjectID: "project", WorkspaceID: "workspace", WorkUnitID: "work", AttemptID: attemptID("work", 1), }, }, }, nil } }, wantCode: BlockerEvidenceRepairDenied, }, } { t.Run(test.name, func(t *testing.T) { harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{ "project": testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown)), }, 1) test.configure(harness) harness.start("project", "workspace", nil) if err := harness.manager.Reconcile(context.Background()); err != nil { t.Fatalf("Reconcile: %v", err) } work := harness.store.snapshot().Projects["project"].Works["work"] if work.Blocker == nil || work.Blocker.Code != test.wantCode { t.Fatalf("work blocker = %#v, want %q", work.Blocker, test.wantCode) } if harness.reviewer.callCount() != 0 { t.Fatalf("review calls = %d, want 0", harness.reviewer.callCount()) } _, repairs := harness.evidence.counts() if repairs != test.wantRepair { t.Fatalf("repair calls = %d, want %d", repairs, test.wantRepair) } }) } t.Run("Pi repair rematches before review", func(t *testing.T) { harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{ "project": testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown)), }, 1) harness.manager.selector = fakeSelector{capacity: 1, providerID: "pi"} harness.invoker.locators["work"] = []LocatorRecord{{ Kind: LocatorSession, Opaque: "native-session", Revision: "session-r1", ProjectID: "project", WorkspaceID: "workspace", WorkUnitID: "work", AttemptID: attemptID("work", 1), }} harness.evidence.observeFunc = func(request WorkflowEvidenceRequest, call int) (ArtifactEvidence, error) { identity := artifactIdentity(request.Project, request.Work, request.Submission) if call == 0 { return ArtifactEvidence{ Active: true, Identity: identity, Completeness: ArtifactPlaceholder, RepairIntent: &EvidenceRepairIntent{ Identity: identity, DispatchOrdinal: request.Work.DispatchOrdinal, NativeLocator: request.Work.Locators[LocatorSession], }, }, nil } return ArtifactEvidence{Active: true, Identity: identity, Completeness: ArtifactComplete}, nil } harness.start("project", "workspace", nil) if err := harness.manager.Reconcile(context.Background()); err != nil { t.Fatalf("Reconcile: %v", err) } observations, repairs := harness.evidence.counts() if observations != 2 || repairs != 1 { t.Fatalf("observe/repair calls = %d/%d, want 2/1", observations, repairs) } if harness.reviewer.callCount() != 1 { t.Fatalf("review calls = %d, want 1 after fresh rematch", harness.reviewer.callCount()) } }) t.Run("Pi repair without a fresh completed match never invokes review", func(t *testing.T) { harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{ "project": testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown)), }, 1) harness.manager.selector = fakeSelector{capacity: 1, providerID: "pi"} harness.invoker.locators["work"] = []LocatorRecord{{ Kind: LocatorSession, Opaque: "native-session", Revision: "session-r1", ProjectID: "project", WorkspaceID: "workspace", WorkUnitID: "work", AttemptID: attemptID("work", 1), }} harness.evidence.observeFunc = func(request WorkflowEvidenceRequest, call int) (ArtifactEvidence, error) { identity := artifactIdentity(request.Project, request.Work, request.Submission) if call == 0 { return ArtifactEvidence{ Active: true, Identity: identity, Completeness: ArtifactPlaceholder, RepairIntent: &EvidenceRepairIntent{ Identity: identity, DispatchOrdinal: request.Work.DispatchOrdinal, NativeLocator: request.Work.Locators[LocatorSession], }, }, nil } return ArtifactEvidence{Active: true, Identity: identity, Completeness: ArtifactPlaceholder}, nil } harness.start("project", "workspace", nil) if err := harness.manager.Reconcile(context.Background()); err != nil { t.Fatalf("Reconcile: %v", err) } work := harness.store.snapshot().Projects["project"].Works["work"] if work.Blocker == nil || work.Blocker.Code != BlockerSubmissionIncomplete { t.Fatalf("work blocker = %#v, want %q", work.Blocker, BlockerSubmissionIncomplete) } observations, repairs := harness.evidence.counts() if observations != 2 || repairs != 1 { t.Fatalf("observe/repair calls = %d/%d, want 2/1", observations, repairs) } if harness.reviewer.callCount() != 0 { t.Fatalf("review calls = %d, want 0 without a fresh completed match", harness.reviewer.callCount()) } }) }