refactor(compass): SEA-1516 collapse the container hop and record agent placement

go/internal/runnerhub/commands.go

@@ -25,9 +25,15 @@ import (
 )
 
 // Provision relays a ProvisionAgentWorkspace command to the owning Runner and
-// returns the container name it created. requestID is the client_request_id
-// idempotency key: a timeout-retry with the same id returns the same container
-// (no duplicate); empty mints a fresh id (no dedup).
+// returns the container name it created plus the id of the Runner that served
+// the call. The Runner id is returned rather than looked up afterwards so the
+// caller records the agent's durable placement against the Runner that ACTUALLY
+// ran the provision — re-reading the registry after the round trip could name a
+// different Runner if one re-enrolled in the meantime, and a placement pointing
+// at the wrong Runner is worse than none (reattach would re-drive the wrong
+// set). requestID is the client_request_id idempotency key: a timeout-retry with
+// the same id returns the same container (no duplicate); empty mints a fresh id
+// (no dedup).
 //
 // The client_request_id alone must NOT be the dedup key: it is a client-chosen
 // string, and two provisions that reuse one value for DIFFERENT workspaces
@@ -39,26 +45,28 @@ import (
 // This mirrors the comms store's (author_account_id, client_request_id)
 // idempotency scoping (store/migrations/0001_init.sql), strengthened to the full
 // provision identity since a provision creates a real isolated container.
-func (h *Hub) Provision(ctx context.Context, requestID string, req *compassv1.ProvisionAgentWorkspaceRequest) (*compassv1.ProvisionAgentWorkspaceResponse, error) {
-	result, err := h.relay(ctx, "", &compassv1internal.SessionsResponse{
+func (h *Hub) Provision(ctx context.Context, requestID string, req *compassv1.ProvisionAgentWorkspaceRequest) (*compassv1.ProvisionAgentWorkspaceResponse, string, error) {
+	result, runnerID, err := h.relay(ctx, "", &compassv1internal.SessionsResponse{
 		RequestId: provisionDedupID(requestID, req),
 		Command:   &compassv1internal.SessionsResponse_Provision{Provision: req},
 	})
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	resp := result.GetProvision()
 	// Record which agent account this container was provisioned for, so a later
 	// Start can promote it to a session binding RelayCommsCall resolves against
 	// (comms-tools design T2). The Runner never asserts this account; it is the
-	// Server's own record, keyed by the container name the Runner returned.
+	// Server's own record, keyed by the container name the Runner returned. This
+	// binding is the LIVE comms binding only, cleared on re-enroll; the DURABLE
+	// container/Runner placement is the caller's store write.
 	h.bindContainer(resp.GetContainerName(), store.AccountID(req.GetAgentAccountId()))
-	return resp, nil
+	return resp, runnerID, nil
 }
 
 // Start relays a StartAgentSession command to the owning Runner.
 func (h *Hub) Start(ctx context.Context, requestID string, req *compassv1.StartAgentSessionRequest) (*compassv1.StartAgentSessionResponse, error) {
-	result, err := h.relay(ctx, req.GetContainerName(), &compassv1internal.SessionsResponse{
+	result, _, err := h.relay(ctx, req.GetContainerName(), &compassv1internal.SessionsResponse{
 		RequestId: orNewRequestID(requestID),
 		Command:   &compassv1internal.SessionsResponse_Start{Start: req},
 	})
@@ -76,7 +84,7 @@ func (h *Hub) Start(ctx context.Context, requestID string, req *compassv1.StartA
 
 // Stop relays a StopAgentSession command to the owning Runner.
 func (h *Hub) Stop(ctx context.Context, requestID string, req *compassv1.StopAgentSessionRequest) (*compassv1.StopAgentSessionResponse, error) {
-	result, err := h.relay(ctx, req.GetSessionId(), &compassv1internal.SessionsResponse{
+	result, _, err := h.relay(ctx, req.GetSessionId(), &compassv1internal.SessionsResponse{
 		RequestId: orNewRequestID(requestID),
 		Command:   &compassv1internal.SessionsResponse_Stop{Stop: req},
 	})
@@ -92,7 +100,7 @@ func (h *Hub) Stop(ctx context.Context, requestID string, req *compassv1.StopAge
 
 // Reload relays a ReloadAgentSession command to the owning Runner.
 func (h *Hub) Reload(ctx context.Context, requestID string, req *compassv1.ReloadAgentSessionRequest) (*compassv1.ReloadAgentSessionResponse, error) {
-	result, err := h.relay(ctx, req.GetSessionId(), &compassv1internal.SessionsResponse{
+	result, _, err := h.relay(ctx, req.GetSessionId(), &compassv1internal.SessionsResponse{
 		RequestId: orNewRequestID(requestID),
 		Command:   &compassv1internal.SessionsResponse_Reload{Reload: req},
 	})
@@ -105,7 +113,7 @@ func (h *Hub) Reload(ctx context.Context, requestID string, req *compassv1.Reloa
 // Status relays a GetAgentStatus command to the owning Runner — the Runner is
 // authoritative for live session truth, so the Server reconciles to its answer.
 func (h *Hub) Status(ctx context.Context, requestID string, req *compassv1.GetAgentStatusRequest) (*compassv1.GetAgentStatusResponse, error) {
-	result, err := h.relay(ctx, req.GetSessionId(), &compassv1internal.SessionsResponse{
+	result, _, err := h.relay(ctx, req.GetSessionId(), &compassv1internal.SessionsResponse{
 		RequestId: orNewRequestID(requestID),
 		Command:   &compassv1internal.SessionsResponse_Status{Status: req},
 	})
@@ -119,20 +127,22 @@ func (h *Hub) Status(ctx context.Context, requestID string, req *compassv1.GetAg
 // the outcome to a Connect status: a RunnerError result becomes the mapped
 // Connect code; a transport failure (no Runner, stream drop) becomes
 // Unavailable. sessionKey selects the owning Runner (single-Runner MVP: any
-// non-empty key resolves the one Runner).
-func (h *Hub) relay(ctx context.Context, sessionKey string, cmd *compassv1internal.SessionsResponse) (*compassv1internal.SessionsRequest, error) {
-	router, err := h.routerFor(sessionKey)
+// non-empty key resolves the one Runner). The served Runner's id is returned
+// alongside the result for the one caller that must attribute the command to a
+// Runner (Provision, recording a durable placement); the rest discard it.
+func (h *Hub) relay(ctx context.Context, sessionKey string, cmd *compassv1internal.SessionsResponse) (*compassv1internal.SessionsRequest, string, error) {
+	router, runnerID, err := h.routerFor(sessionKey)
 	if err != nil {
-		return nil, connect.NewError(connect.CodeUnavailable, err)
+		return nil, "", connect.NewError(connect.CodeUnavailable, err)
 	}
 	result, err := router.dispatch(ctx, cmd)
 	if err != nil {
-		return nil, connect.NewError(connect.CodeUnavailable, err)
+		return nil, "", connect.NewError(connect.CodeUnavailable, err)
 	}
 	if runnerErr := result.GetError(); runnerErr != nil {
-		return nil, runnerErrorToConnect(runnerErr)
+		return nil, "", runnerErrorToConnect(runnerErr)
 	}
-	return result, nil
+	return result, runnerID, nil
 }
 
 // runnerErrorToConnect maps a RunnerError to the Connect status the client sees.

go/internal/runnerhub/commands_test.go

@@ -57,7 +57,7 @@ func TestCommandsNoRunnerIsUnavailable(t *testing.T) {
 		call func() error
 	}{
 		{"provision", func() error {
-			_, err := hub.Provision(ctx, "r1", &compassv1.ProvisionAgentWorkspaceRequest{})
+			_, _, err := hub.Provision(ctx, "r1", &compassv1.ProvisionAgentWorkspaceRequest{})
 			return err
 		}},
 		{"start", func() error {
@@ -99,7 +99,7 @@ func TestStartRelaySurfacesAlreadyRunningAsAlreadyExists(t *testing.T) {
 	// Enroll a Runner and bind a send that answers every command with an
 	// ALREADY_RUNNING error result correlated by the pushed request id.
 	hub.enroll("runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"})
-	router, err := hub.routerFor("any")
+	router, _, err := hub.routerFor("any")
 	if err != nil {
 		t.Fatalf("routerFor after enroll = %v, want a router", err)
 	}
@@ -130,7 +130,7 @@ func TestStartRelaySurfacesAlreadyRunningAsAlreadyExists(t *testing.T) {
 func TestStartRelayReturnsSessionIdOnSuccess(t *testing.T) {
 	hub := newHubOnly()
 	hub.enroll("runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"})
-	router, _ := hub.routerFor("any")
+	router, _, _ := hub.routerFor("any")
 	router.attach(func(cmd *compassv1internal.SessionsResponse) error {
 		go router.complete(&compassv1internal.SessionsRequest{
 			RequestId: cmd.GetRequestId(),

go/internal/runnerhub/concurrent_dispatch_test.go

@@ -59,7 +59,7 @@ func TestConcurrentDispatchOverRealStreamNoDataRace(t *testing.T) {
 	// before router.attach ran would (correctly) find no live send.
 	waitRouterAttached(t, hub)
 
-	router, err := hub.routerFor("runner-1")
+	router, _, err := hub.routerFor("runner-1")
 	if err != nil {
 		t.Fatalf("routerFor after enroll+attach = %v, want the live router", err)
 	}

go/internal/runnerhub/handler.go

@@ -70,7 +70,7 @@ func (h *Handler) Sessions(ctx context.Context, stream *connect.BidiStream[compa
 	if !ok {
 		return errUnauthenticated
 	}
-	router, err := h.hub.routerFor(subj.ID)
+	router, _, err := h.hub.routerFor(subj.ID)
 	if err != nil {
 		// A Sessions stream with no enrolled Runner — the Runner must Enroll
 		// before opening Sessions.

go/internal/runnerhub/helpers_test.go

@@ -51,7 +51,7 @@ func waitRouterAttached(t *testing.T, hub *Hub) {
 			t.Fatal("runner command router never attached a live Sessions send")
 		default:
 		}
-		if router, err := hub.routerFor("gate"); err == nil {
+		if router, _, err := hub.routerFor("gate"); err == nil {
 			router.mu.Lock()
 			attached := router.send != nil
 			router.mu.Unlock()

go/internal/runnerhub/hub.go

@@ -262,13 +262,17 @@ func (h *Hub) enroll(id string, subject store.Subject) (reattached bool) {
 	return reattached
 }
 
-// router returns the attached Runner's command router, or an error when no
-// Runner is enrolled (a session command with no Runner to serve it).
-func (h *Hub) routerFor(sessionID string) (*commandRouter, error) {
+// routerFor returns the attached Runner's command router and its id, or an error
+// when no Runner is enrolled (a session command with no Runner to serve it). The
+// id travels out with the router so a caller that must attribute the call to a
+// Runner (Provision, recording an agent's durable placement) names the Runner
+// that actually served it, rather than re-reading the registry afterwards and
+// racing a re-enroll onto the wrong id.
+func (h *Hub) routerFor(sessionID string) (*commandRouter, string, error) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	if h.runner == nil {
-		return nil, fmt.Errorf("no runner enrolled to serve session %q", sessionID)
+		return nil, "", fmt.Errorf("no runner enrolled to serve session %q", sessionID)
 	}
-	return h.runner.router, nil
+	return h.runner.router, h.runner.id, nil
 }

go/internal/runnerhub/hub_test.go

@@ -269,7 +269,7 @@ func TestEnrollDuplicateReattaches(t *testing.T) {
 	}
 	// A router is resolvable after enrollment (a session command has a Runner to
 	// serve it).
-	if _, err := hub.routerFor("any"); err != nil {
+	if _, _, err := hub.routerFor("any"); err != nil {
 		t.Fatalf("routerFor after enroll = %v, want a live router", err)
 	}
 }

go/internal/runnerhub/integration_pgtest_test.go

@@ -212,7 +212,7 @@ func provisionWhenSeamLive(t *testing.T, ctx context.Context, hub *runnerhub.Hub
 	t.Helper()
 	deadline := time.After(integrationTimeout)
 	for {
-		resp, err := hub.Provision(ctx, "prov-1", &compassv1.ProvisionAgentWorkspaceRequest{
+		resp, _, err := hub.Provision(ctx, "prov-1", &compassv1.ProvisionAgentWorkspaceRequest{
 			AgentAccountId: string(agentID),
 			Repo:           &compassv1.ProvisionAgentWorkspaceRequest_LocalPath{LocalPath: "/mirror/repo.git"},
 		})

go/internal/runnerhub/provision_dedup_test.go

@@ -53,7 +53,7 @@ type provisionOutcome struct {
 func enrollAttached(t *testing.T, hub *Hub, send *recordingSend) *commandRouter {
 	t.Helper()
 	hub.enroll("runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"})
-	router, err := hub.routerFor("any")
+	router, _, err := hub.routerFor("any")
 	if err != nil {
 		t.Fatalf("routerFor after enroll = %v, want the live router", err)
 	}
@@ -83,7 +83,7 @@ func TestProvisionSameClientRequestIdDedups(t *testing.T) {
 		}
 		outcomes := make(chan provisionOutcome, 2)
 		call := func() {
-			resp, err := hub.Provision(context.Background(), id, req)
+			resp, _, err := hub.Provision(context.Background(), id, req)
 			outcomes <- provisionOutcome{resp, err}
 		}
 
@@ -140,7 +140,7 @@ func TestProvisionEmptyClientRequestIdDoesNotDedup(t *testing.T) {
 		req := &compassv1.ProvisionAgentWorkspaceRequest{}
 		outcomes := make(chan provisionOutcome, 2)
 		call := func() {
-			resp, err := hub.Provision(context.Background(), "", req)
+			resp, _, err := hub.Provision(context.Background(), "", req)
 			outcomes <- provisionOutcome{resp, err}
 		}
 
@@ -211,7 +211,7 @@ func TestProvisionSameIdDifferentWorkspaceDoesNotDedup(t *testing.T) {
 		}
 		outcomes := make(chan provisionOutcome, 2)
 		call := func(req *compassv1.ProvisionAgentWorkspaceRequest) {
-			resp, err := hub.Provision(context.Background(), id, req)
+			resp, _, err := hub.Provision(context.Background(), id, req)
 			outcomes <- provisionOutcome{resp, err}
 		}
 

go/internal/runnerhub/relay_comms_test.go

@@ -331,7 +331,7 @@ func TestRelayCommsCallStoppedSessionFailsClosedNotFound(t *testing.T) {
 func TestProvisionThenStartBindsSessionToProvisionedAccount(t *testing.T) {
 	hub := newHubOnly()
 	hub.enroll("runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"})
-	router, err := hub.routerFor("any")
+	router, _, err := hub.routerFor("any")
 	if err != nil {
 		t.Fatalf("routerFor after enroll = %v, want a router", err)
 	}
@@ -356,7 +356,7 @@ func TestProvisionThenStartBindsSessionToProvisionedAccount(t *testing.T) {
 	})
 
 	ctx := context.Background()
-	if _, err := hub.Provision(ctx, "req-prov", &compassv1.ProvisionAgentWorkspaceRequest{AgentAccountId: "acct-agent"}); err != nil {
+	if _, _, err := hub.Provision(ctx, "req-prov", &compassv1.ProvisionAgentWorkspaceRequest{AgentAccountId: "acct-agent"}); err != nil {
 		t.Fatalf("Provision = %v, want success", err)
 	}
 	if _, err := hub.Start(ctx, "req-start", &compassv1.StartAgentSessionRequest{ContainerName: "cont-1"}); err != nil {
@@ -378,7 +378,7 @@ func TestProvisionThenStartBindsSessionToProvisionedAccount(t *testing.T) {
 func TestProvisionWithEmptyAccountLeavesNoBindingAndFailsClosed(t *testing.T) {
 	hub, comms := newHubWithComms()
 	hub.enroll("runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"})
-	router, err := hub.routerFor("any")
+	router, _, err := hub.routerFor("any")
 	if err != nil {
 		t.Fatalf("routerFor after enroll = %v, want a router", err)
 	}
@@ -401,7 +401,7 @@ func TestProvisionWithEmptyAccountLeavesNoBindingAndFailsClosed(t *testing.T) {
 	})
 
 	ctx := context.Background()
-	if _, err := hub.Provision(ctx, "req-prov", &compassv1.ProvisionAgentWorkspaceRequest{AgentAccountId: ""}); err != nil {
+	if _, _, err := hub.Provision(ctx, "req-prov", &compassv1.ProvisionAgentWorkspaceRequest{AgentAccountId: ""}); err != nil {
 		t.Fatalf("Provision (empty account) = %v, want success", err)
 	}
 	if _, err := hub.Start(ctx, "req-start", &compassv1.StartAgentSessionRequest{ContainerName: "cont-1"}); err != nil {