orchestra/internal/integration/acefca_test.go

package integration

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

	"orchestra/internal/agentctx"
	"orchestra/internal/authz"
	"orchestra/internal/continuity"
	"orchestra/internal/domain"
	"orchestra/internal/human"
	"orchestra/internal/operations"
	"orchestra/internal/registry"
	"orchestra/internal/review"
	"orchestra/internal/router"
	"orchestra/internal/store"
	"orchestra/internal/workphase"
)

// phaseSource feeds one human comment at a time, in the order the test wants
// them observed.
type phaseSource struct {
	pending []human.Input
	served  int
}

func (p *phaseSource) FetchAfter(_ context.Context, task domain.Task, _ store.SourceCursor) ([]human.Input, store.SourceCursor, error) {
	if p.served >= len(p.pending) {
		return nil, store.SourceCursor{}, nil
	}
	in := p.pending[:p.served+1]
	p.served++
	return in, store.SourceCursor{TaskID: task.ID, Provider: "gitea", Cursor: in[len(in)-1].ExternalID}, nil
}

// The whole foundation in one test: research discovers A, the human corrects
// to B, the phase rotates, planning proposes C, the human rejects C for D,
// implementation starts from D, and the handoff still full of A and C material
// must read as history only.
func TestPhaseBoundariesCarryHumanAuthorityAndDemoteHandoffs(t *testing.T) {
	s, reg, _ := setup(t)
	task := ingest(t, s, "381")
	project, ok := reg.Project("p")
	if !ok {
		t.Fatal("project missing")
	}

	src := &phaseSource{pending: []human.Input{
		{Provider: "gitea", ExternalID: "918", Author: "kami", Body: "no, aggregate per person, not per figure"},
		{Provider: "gitea", ExternalID: "919", Author: "kami", Body: "reject the cache, add the index instead"},
	}}
	rec := &human.Reconciler{Store: s, Sources: map[string]human.Source{"gitea": src}}
	s.PreLease = func(id string) error { return rec.Reconcile(context.Background(), id) }

	// Store.Lease is the choke point every launch path goes through, so the
	// session loop drives it directly and renders what a launch would render.
	prompts := map[domain.WorkPhase]string{}
	onLease := func(taskID string) {
		current, _ := s.Task(taskID)
		prompts[phaseOf(current)] = buildFor(t, s, current)
	}

	// Frame. The first lease reconciles nothing yet.
	leaseAndRelease(t, s, onLease, task.ID, nil)
	advance(t, s, project, task.ID, nil)

	// Research. The human's correction lands before the research session starts.
	leaseAndRelease(t, s, onLease, task.ID, &continuity.Handoff{
		Meta:   continuity.Meta{ID: "h-research", Reason: "threshold"},
		Anchor: anchor(),
		Action: "keep aggregating per figure",
	})
	if got := prompts[domain.WorkPhaseResearch]; !strings.Contains(got, "aggregate per person") {
		t.Fatalf("research context missing the correction:\n%s", got)
	}

	// Research seals its findings, including the discovery the human corrected.
	advance(t, s, project, task.ID, sealedResearch(t))

	// Plan. It receives the sealed research and the standing correction, and
	// the previous handoff must be demoted to history.
	leaseAndRelease(t, s, onLease, task.ID, &continuity.Handoff{
		Meta:      continuity.Meta{ID: "h-plan", Reason: "threshold"},
		Anchor:    anchor(),
		Action:    "add the cache",
		Remaining: []string{"finish per-figure aggregation"},
	})
	planCtx := prompts[domain.WorkPhasePlan]
	assertContains(t, "plan", planCtx, "## Accepted research", "aggregate per person", "runs per figure")
	assertOrder(t, "plan", planCtx, "aggregate per person", "## Continuity from the previous session", "keep aggregating per figure")

	advance(t, s, project, task.ID, sealedPlan(t))

	// Implement. The second correction rejects the plan's approach. It must be
	// standing in the implementation context, above both the sealed plan and
	// the handoff that still names the rejected approach.
	leaseAndRelease(t, s, onLease, task.ID, &continuity.Handoff{
		Meta:   continuity.Meta{ID: "h-impl", Reason: "threshold"},
		Anchor: anchor(),
		Action: "add the cache",
	})
	implCtx := prompts[domain.WorkPhaseImplement]
	assertContains(t, "implement",
		implCtx,
		"## Accepted research",
		"## Accepted plan",
		"aggregate per person",
		"add the cache",
		"reject the cache, add the index instead",
	)
	// Both corrections outrank both sealed artifacts and the handoff.
	assertOrder(t, "implement", implCtx,
		"## Current human decisions",
		"reject the cache",
		"## Accepted research",
		"## Accepted plan",
		"## Continuity from the previous session",
		"finish per-figure aggregation",
	)
	if !strings.Contains(implCtx, "History, not instruction.") {
		t.Fatal("handoff not demoted in the implementation context")
	}

	// Review sees the plan and the decisions, never the research transcript.
	advance(t, s, project, task.ID, nil)
	leaseAndRelease(t, s, onLease, task.ID, nil)
	reviewCtx := prompts[domain.WorkPhaseReview]
	assertContains(t, "review", reviewCtx, "## Accepted plan", "reject the cache")
	if strings.Contains(reviewCtx, "## Accepted research") {
		t.Fatalf("review received the research artifact:\n%s", reviewCtx)
	}
}

func anchor() continuity.Anchor {
	return continuity.Anchor{GitSHA: "0123456789012345678901234567890123456789", Branch: "orchestra/task"}
}

func phaseOf(t domain.Task) domain.WorkPhase {
	if t.WorkPhase == "" {
		return domain.WorkPhaseFrame
	}
	return t.WorkPhase
}

// buildFor renders exactly what a launch would render, from the store alone.
func buildFor(t *testing.T, s *store.Store, task domain.Task) string {
	t.Helper()
	intent, err := s.EffectiveIntent(task.ID)
	if err != nil {
		t.Fatal(err)
	}
	in := agentctx.Input{Task: task, Intent: intent, Phase: phaseOf(task), DecisionRequest: task.DecisionRequest, Git: agentctx.GitState{Worktree: "/srv/wt", Branch: "orchestra/" + task.ID}}
	if task.HandoffRef != "" {
		h, err := continuity.Load(task.HandoffRef, s)
		if err != nil {
			t.Fatal(err)
		}
		in.Handoff = &h
	}
	if task.ResearchRef != "" {
		b, err := s.Artifact(task.ResearchRef)
		if err != nil {
			t.Fatal(err)
		}
		r, err := workphase.DecodeResearch(b)
		if err != nil {
			t.Fatal(err)
		}
		in.Research = &r
	}
	if task.PlanRef != "" {
		b, err := s.Artifact(task.PlanRef)
		if err != nil {
			t.Fatal(err)
		}
		p, err := workphase.DecodePlan(b)
		if err != nil {
			t.Fatal(err)
		}
		in.Plan = &p
	}
	built, err := agentctx.Build(in)
	if err != nil {
		t.Fatal(err)
	}
	return built.System + "\n\n" + built.Task
}

// leaseAndRelease runs one session: the router mints the lease, which
// reconciles human input first, then the session releases with the handoff it
// wrote (when it wrote one).
func leaseAndRelease(t *testing.T, s *store.Store, onLease func(string), taskID string, h *continuity.Handoff) {
	t.Helper()
	if _, err := s.Lease(taskID, "h1", 30*time.Minute); err != nil {
		t.Fatal(err)
	}
	onLease(taskID)
	task, _ := s.Task(taskID)
	payload := map[string]any{
		"harness_id": task.Lease.HarnessID, "lease_epoch": task.Lease.Epoch,
		"expected_version": task.Version, "reason": "threshold",
	}
	if h != nil {
		b, err := continuity.Encode(*h)
		if err != nil {
			t.Fatal(err)
		}
		ref, err := s.PutArtifact(b)
		if err != nil {
			t.Fatal(err)
		}
		payload["handoff_ref"] = ref
		payload["anchor_sha"] = "0123456789012345678901234567890123456789"
	}
	if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: taskID, Version: task.Version + 1, Surface: string(authz.System), Payload: mustJSON(payload)}); err != nil {
		t.Fatal(err)
	}
}

func advance(t *testing.T, s *store.Store, project registry.Project, taskID string, artifact []byte) {
	t.Helper()
	if _, err := operations.AdvanceWorkPhase(s, project, taskID, artifact); err != nil {
		t.Fatal(err)
	}
}

func sealedResearch(t *testing.T) []byte {
	t.Helper()
	b, err := workphase.Encode(workphase.Research{
		Findings: []workphase.Finding{{ID: "r1", Confidence: workphase.Fact, Claim: "attribution runs per figure", Evidence: "internal/attr/attr.go:88"}},
		DeadEnds: []workphase.DeadEnd{{Tried: "figure plurality", WhyFailed: "no measured gain"}},
	})
	if err != nil {
		t.Fatal(err)
	}
	return b
}

func sealedPlan(t *testing.T) []byte {
	t.Helper()
	b, err := workphase.Encode(workphase.Plan{
		Changes:      []workphase.Change{{Target: "internal/attr/attr.go", Intent: "add the cache"}},
		Verification: []string{"go test ./internal/attr/"},
	})
	if err != nil {
		t.Fatal(err)
	}
	return b
}

func assertContains(t *testing.T, phase, ctx string, want ...string) {
	t.Helper()
	for _, w := range want {
		if !strings.Contains(ctx, w) {
			t.Fatalf("%s context missing %q:\n%s", phase, w, ctx)
		}
	}
}

func assertOrder(t *testing.T, phase, ctx string, seq ...string) {
	t.Helper()
	last := -1
	for _, s := range seq {
		at := strings.Index(ctx, s)
		if at < 0 {
			t.Fatalf("%s context missing %q:\n%s", phase, s, ctx)
		}
		if at < last {
			t.Fatalf("%s context has %q out of order:\n%s", phase, s, ctx)
		}
		last = at
	}
}

// The trajectory gate proof. Research finds A and the plan proposes B. The
// human keeps A, rejects B, and chooses C. The plan stays sealed as historical
// evidence, and the implementation context opens with C above it.
func TestTrajectoryGateCorrectionOutranksTheSealedPlan(t *testing.T) {
	s, reg, _ := setup(t)
	task := ingest(t, s, "381")
	base, _ := reg.Project("p")
	project := base
	project.TrajectoryGate = map[string]string{"plan_to_implement": "required"}

	src := &phaseSource{}
	rec := &human.Reconciler{Store: s, Sources: map[string]human.Source{"gitea": src}}
	s.PreLease = func(id string) error { return rec.Reconcile(context.Background(), id) }

	// frame -> research -> plan, sealing the research that established A.
	advance(t, s, project, task.ID, nil)
	advance(t, s, project, task.ID, sealedResearch(t))

	// plan -> implement proposes B and stops at the gate.
	proposal := sealedPlan(t)
	if _, err := operations.AdvanceWorkPhase(s, project, task.ID, proposal); !errors.Is(err, operations.ErrTrajectoryGate) {
		t.Fatalf("want the gate to stop this, got %v", err)
	}
	blocked, _ := s.Task(task.ID)
	if !strings.Contains(blocked.Blocker, "add the cache") {
		t.Fatalf("packet does not carry the proposal:\n%s", blocked.Blocker)
	}

	// The human answers through the ordinary comment path.
	src.pending = []human.Input{{
		Provider: "gitea", ExternalID: "918", Author: "kami",
		Body: "keep the per-figure finding, but do not add the cache, add the index instead",
	}}
	if err := rec.Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}

	// Now the gate opens and the plan seals.
	if _, err := operations.AdvanceWorkPhase(s, project, task.ID, proposal); err != nil {
		t.Fatal(err)
	}
	got, _ := s.Task(task.ID)
	if got.WorkPhase != domain.WorkPhaseImplement {
		t.Fatalf("phase = %q", got.WorkPhase)
	}
	if got.PlanRef == "" {
		t.Fatal("the rejected plan must stay sealed as evidence")
	}

	// The implementation context: the correction first, the sealed plan below
	// it, and the rejected approach still visible as what was proposed.
	ctx := buildFor(t, s, got)
	assertContains(t, "implement", ctx, "add the index instead", "## Accepted plan")
	// The plan's own line, not the decision's mention of it: the rejected
	// approach stays visible as evidence, below the correction that rejected it.
	assertOrder(t, "implement", ctx,
		"## Current human decisions",
		"add the index instead",
		"## Accepted plan",
		"internal/attr/attr.go: add the cache",
	)
}

// The grilling proof. The plan says add the cache. Implementation discovers a
// compatibility requirement the repository does not settle, so the task stops
// with one bounded question. The human answers in ordinary prose, the task
// resumes, and the answer leads the implementation context with the rejected
// plan still below it.
func TestBlockingQuestionResumesWithTheAnswerOnTop(t *testing.T) {
	s, reg, _ := setup(t)
	task := ingest(t, s, "381")
	project, _ := reg.Project("p")

	src := &phaseSource{}
	rec := &human.Reconciler{Store: s, Sources: map[string]human.Source{"gitea": src}}
	s.PreLease = func(id string) error { return rec.Reconcile(context.Background(), id) }

	advance(t, s, project, task.ID, nil)
	advance(t, s, project, task.ID, sealedResearch(t))
	advance(t, s, project, task.ID, sealedPlan(t))
	if got, _ := s.Task(task.ID); got.WorkPhase != domain.WorkPhaseImplement {
		t.Fatalf("phase = %q", got.WorkPhase)
	}

	// A question comes from the session that owns the task, and Store.Append
	// fences it on that lease.
	if _, err := s.Lease(task.ID, "h1", time.Hour); err != nil {
		t.Fatal(err)
	}

	// Implementation hits genuine ambiguity and asks once.
	if _, err := operations.RequestHumanDecision(s, project, task.ID, domain.DecisionRequest{
		Question: "must the old cache contract stay compatible?",
		Why:      "the accepted plan adds a cache, and two callers depend on behaviour the repository never documents",
		Options: []domain.DecisionOption{
			{ID: "preserve", Description: "keep the contract", Tradeoff: "larger change"},
			{ID: "break", Description: "change it", Tradeoff: "two callers must migrate"},
		},
		Evidence: []string{"internal/attr/attr.go:88 documents neither behaviour"},
	}); err != nil {
		t.Fatal(err)
	}
	blocked, _ := s.Task(task.ID)
	if blocked.State != domain.StateBlocked || blocked.BlockReason != domain.BlockReasonHumanDecision {
		t.Fatalf("task = %+v", blocked)
	}
	// While blocked, the question is in the context exactly once and marked as
	// the human's to answer.
	waiting := buildFor(t, s, blocked)
	assertContains(t, "blocked", waiting, "## Human decision required", "must the old cache contract stay compatible?", "Do not answer it yourself")

	// The human answers through the ordinary comment path, in their own words.
	src.pending = []human.Input{{
		Provider: "gitea", ExternalID: "918", Author: "kami",
		Body: "break compatibility; update the two callers",
	}}
	if err := rec.Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := operations.ResumeAnsweredBlockers(s); err != nil {
		t.Fatal(err)
	}
	resumed, _ := s.Task(task.ID)
	if resumed.State != domain.StateQueued {
		t.Fatalf("state = %s", resumed.State)
	}

	// The resumed context: the answer above the sealed plan, and the question
	// gone because it is answered.
	ctx := buildFor(t, s, resumed)
	assertContains(t, "resumed", ctx, "break compatibility", "## Accepted plan", "internal/attr/attr.go: add the cache")
	if strings.Contains(ctx, "## Human decision required") {
		t.Fatalf("an answered question is still being asked:\n%s", ctx)
	}
	assertOrder(t, "resumed", ctx,
		"## Current human decisions",
		"break compatibility",
		"## Accepted research",
		"## Accepted plan",
		"internal/attr/attr.go: add the cache",
	)
}

// The independent review proof. The plan says cache, the human corrected to
// index, and the implementation used index. The reviewer sees the correction,
// the plan as subordinate evidence, and the exact diff. It sees nothing the
// implementation session said about its own work.
func TestReviewSessionIsIndependent(t *testing.T) {
	s, reg, _ := setup(t)
	task := ingest(t, s, "381")
	project, _ := reg.Project("p")
	project.QualityGate = "go test ./..."

	src := &phaseSource{pending: []human.Input{{
		Provider: "gitea", ExternalID: "918", Author: "kami",
		Body: "do not add the cache, add the index instead",
	}}}
	rec := &human.Reconciler{Store: s, Sources: map[string]human.Source{"gitea": src}}
	s.PreLease = func(id string) error { return rec.Reconcile(context.Background(), id) }

	advance(t, s, project, task.ID, nil)
	advance(t, s, project, task.ID, sealedResearch(t))
	advance(t, s, project, task.ID, sealedPlan(t))

	// The correction arrives, and an implementation session leaves a handoff
	// full of its own account of the work.
	if err := rec.Reconcile(context.Background(), task.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := s.Lease(task.ID, "h1", 30*time.Minute); err != nil {
		t.Fatal(err)
	}
	leased, _ := s.Task(task.ID)
	handoff := continuity.Handoff{
		Meta:      continuity.Meta{ID: "h-impl", Reason: "milestone"},
		Anchor:    anchor(),
		Action:    "finish the index lookup",
		Remaining: []string{"the implementation believes this is correct"},
	}
	encoded, err := continuity.Encode(handoff)
	if err != nil {
		t.Fatal(err)
	}
	ref, err := s.PutArtifact(encoded)
	if err != nil {
		t.Fatal(err)
	}
	if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: leased.Version + 1, Surface: string(authz.System), Payload: mustJSON(map[string]any{
		"handoff_ref": ref, "anchor_sha": anchor().GitSHA, "reason": "milestone",
		"harness_id": leased.Lease.HarnessID, "lease_epoch": leased.Lease.Epoch, "expected_version": leased.Version,
	})}); err != nil {
		t.Fatal(err)
	}

	ev := review.Evidence{
		BaseSHA: "0000000000000000000000000000000000000000", ResultSHA: shaImpl,
		Diff:        "--- a/internal/attr/attr.go\n+++ b/internal/attr/attr.go\n-\tcache.Get(id)\n+\tindex.Lookup(id)\n",
		GateCommand: "go test ./...", GateExit: 0, GateOutput: "ok\torchestra/internal/attr\t0.02s",
	}
	if _, err := operations.EnterReview(s, project, task.ID, ev); err != nil {
		t.Fatal(err)
	}

	// The reviewing session's context.
	reviewing, _ := s.Task(task.ID)
	intent, err := s.EffectiveIntent(task.ID)
	if err != nil {
		t.Fatal(err)
	}
	planArtifact, err := s.Artifact(reviewing.PlanRef)
	if err != nil {
		t.Fatal(err)
	}
	sealedPlanValue, err := workphase.DecodePlan(planArtifact)
	if err != nil {
		t.Fatal(err)
	}
	loaded, err := continuity.Load(reviewing.HandoffRef, s)
	if err != nil {
		t.Fatal(err)
	}
	built, err := agentctx.Build(agentctx.Input{
		Task: reviewing, Intent: intent, Phase: domain.WorkPhaseReview,
		Plan: &sealedPlanValue, Evidence: &ev,
		// Deliberately supplied: Build must refuse to render it in review.
		Handoff: &loaded,
		Git:     agentctx.GitState{Worktree: "/srv/wt", Branch: "orchestra/" + task.ID, HeadSHA: shaImpl},
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx := built.System + "\n\n" + built.Task

	assertContains(t, "review", ctx,
		"add the index instead",
		"## Accepted plan",
		"index.Lookup(id)",
		"quality gate: `go test ./...` exited 0",
		"## Review instructions",
		"Do not redesign the solution",
	)
	// No implementation continuity, and no research either.
	for _, forbidden := range []string{
		"## Continuity from the previous session",
		"the implementation believes this is correct",
		"finish the index lookup",
		"## Accepted research",
	} {
		if strings.Contains(ctx, forbidden) {
			t.Fatalf("review context leaked %q:\n%s", forbidden, ctx)
		}
	}
	// Authority order holds: the correction above the plan, the plan above the
	// diff it produced.
	assertOrder(t, "review", ctx,
		"## Current human decisions",
		"add the index instead",
		"## Accepted plan",
		"## Verified change",
		"index.Lookup(id)",
	)

	// A blocking finding sends the work back with the finding in context.
	if _, err := operations.RecordReview(s, project, task.ID, review.Result{
		ResultSHA: shaImpl,
		Findings: []review.Finding{{
			ID: "f1", Severity: review.Blocker, File: "internal/attr/attr.go", Line: 42,
			Claim: "a stale lease can still enter this branch", Evidence: "no epoch check before the lookup",
		}},
	}); err != nil {
		t.Fatal(err)
	}
	back, _ := s.Task(task.ID)
	if back.WorkPhase != domain.WorkPhaseImplement {
		t.Fatalf("phase = %q, want implement", back.WorkPhase)
	}
	if back.ReviewSatisfied(shaImpl) {
		t.Fatal("a blocker must not satisfy completion")
	}
	findings, err := operations.TaskReview(s, back)
	if err != nil {
		t.Fatal(err)
	}
	fixCtx, err := agentctx.Build(agentctx.Input{
		Task: back, Intent: intent, Phase: domain.WorkPhaseImplement,
		Plan: &sealedPlanValue, Review: findings,
		Git: agentctx.GitState{Worktree: "/srv/wt", Branch: "orchestra/" + task.ID, HeadSHA: shaImpl},
	})
	if err != nil {
		t.Fatal(err)
	}
	assertContains(t, "fix", fixCtx.Task, "## Review findings", "stale lease can still enter this branch", "internal/attr/attr.go:42")
	assertOrder(t, "fix", fixCtx.Task,
		"## Current human decisions",
		"add the index instead",
		"## Accepted plan",
		"## Review findings",
	)
}

const shaImpl = "1111111111111111111111111111111111111111"

// Submission is the end of the agent's involvement and not the end of the
// task. An in-review task is invisible to the router, so no session can pick
// it up while the human holds it.
func TestSubmittedTaskIsNotReassigned(t *testing.T) {
	s, reg, _ := setup(t)
	task := ingest(t, s, "381")
	project, _ := reg.Project("p")
	project.QualityGate = "go test ./..."

	advance(t, s, project, task.ID, nil)
	advance(t, s, project, task.ID, sealedResearch(t))
	advance(t, s, project, task.ID, sealedPlan(t))
	ev := review.Evidence{
		BaseSHA: "0000000000000000000000000000000000000000", ResultSHA: shaImpl,
		Diff: "+ index.Lookup(id)\n", GateCommand: "go test ./...", GateExit: 0,
	}
	if _, err := operations.EnterReview(s, project, task.ID, ev); err != nil {
		t.Fatal(err)
	}
	if _, err := operations.RecordReview(s, project, task.ID, review.Result{ResultSHA: shaImpl}); err != nil {
		t.Fatal(err)
	}

	plan, err := operations.PrepareSubmission(s, project, task.ID, shaImpl,
		domain.GateResult{Command: "go test ./...", SHA: shaImpl}, operations.Notes{})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := operations.ExecuteSubmission(context.Background(), s, plan, submitTo("142"), func(context.Context) (string, error) { return shaImpl, nil }); err != nil {
		t.Fatal(err)
	}

	got, _ := s.Task(task.ID)
	if got.State != domain.StateInReview {
		t.Fatalf("state = %s, want in_review", got.State)
	}
	if got.State == domain.StateCompleted {
		t.Fatal("submission must not complete the task")
	}
	rt := router.Router{Store: s, Registry: reg, Reachability: alwaysReachable{}}
	leased, err := rt.AssignPending()
	if err != nil {
		t.Fatal(err)
	}
	if len(leased) != 0 {
		t.Fatalf("an in-review task was reassigned: %v", leased)
	}
	if _, err := s.Lease(task.ID, "h1", time.Minute); err == nil {
		t.Fatal("an in-review task must not be leasable")
	}
}

type stubPublisher struct{ id string }

func (p stubPublisher) Push(_ context.Context, _, _, sha string) (string, error) { return sha, nil }
func (p stubPublisher) EnsurePR(context.Context, operations.SubmissionPlan) (domain.ExternalRef, error) {
	return domain.ExternalRef{Provider: "gitea:p", ID: p.id, URL: "https://git/pulls/" + p.id}, nil
}

func submitTo(id string) operations.Publisher { return stubPublisher{id: id} }