From 28f241ff7893742359acc7dfbd6e2648c1625324 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 12 Sep 2026 13:13:20 -0400 Subject: [PATCH 1/2] feat(runner): declare the runtime tier and egress posture at enrollment The Runner stamped both fields in its own Status path, which no render surface reads: the CLI reads the board snapshot and the UI reads the lifecycle bus, and each built its own AgentSessionStatus carrying only the session, state and account. Every real session therefore reported an unknown tier and posture while the unit tests passed, because each test exercised a seam that already had the fields set. Tier and posture are facts of the one backend a Runner drives, not of a session, so EnrollRequest carries them once and the hub stamps them onto every status it publishes for that Runner. The relayed SessionFrame stays unchanged. Reading them under the hub's lock means a status published after a reattach reflects the newly enrolled Runner, never the previous one, and an absent Runner yields UNSPECIFIED rather than a plausible default. PostureOf joins TierOf as an engine probe so the value declared at enrollment is derived the same way a live workload derives it; AgentRuntime.EgressPosture now delegates to it rather than repeating the marker check. Co-authored-by: Matt Wilkinson --- go/internal/board/projection.go | 32 ++- go/internal/gen/compass/v1/runner.pb.go | 197 ++++++++++-------- go/internal/runner/enroll_identity_test.go | 92 ++++++++ go/internal/runner/runner.go | 4 +- go/internal/runnerhub/binding_cache_test.go | 31 +-- go/internal/runnerhub/commands_test.go | 12 +- go/internal/runnerhub/config_fetch_test.go | 13 +- go/internal/runnerhub/config_signal_test.go | 7 +- go/internal/runnerhub/deliveryarm_test.go | 4 +- go/internal/runnerhub/enroll_reap_test.go | 9 +- go/internal/runnerhub/handler.go | 2 +- go/internal/runnerhub/hub.go | 34 ++- go/internal/runnerhub/hub_test.go | 4 +- go/internal/runnerhub/provision_dedup_test.go | 2 +- go/internal/runnerhub/relay_comms_test.go | 16 +- .../runnerhub/relay_operator_fault_test.go | 4 +- .../runnerhub/runtime_identity_test.go | 181 ++++++++++++++++ go/internal/runnerhub/secrets_test.go | 21 +- go/internal/runtime/agent.go | 5 +- go/internal/runtime/tier.go | 12 ++ proto/compass/v1/runner.proto | 7 + 21 files changed, 525 insertions(+), 164 deletions(-) create mode 100644 go/internal/runner/enroll_identity_test.go create mode 100644 go/internal/runnerhub/runtime_identity_test.go diff --git a/go/internal/board/projection.go b/go/internal/board/projection.go index 5cc67f9a2..ca33df362 100644 --- a/go/internal/board/projection.go +++ b/go/internal/board/projection.go @@ -50,12 +50,16 @@ type Projection struct { sessions map[string]sessionEntry } -// sessionEntry is the board's per-session record: the latest state and the -// agent account it was attributed to (empty when the hub could not resolve the -// binding — the stated DL-167 residual gap). +// sessionEntry is the board's per-session record: the latest state, the agent +// account it was attributed to (empty when the hub could not resolve the +// binding — the stated DL-167 residual gap), and the owning Runner's runtime +// tier and egress posture (stamped from enrollment, so the snapshot path carries +// them too). type sessionEntry struct { - state compassv1.AgentSessionState - account string + state compassv1.AgentSessionState + account string + tier compassv1.RuntimeTier + egressPosture compassv1.EgressPosture } // NewProjection constructs an empty board over the SubscribeEvents bus it fans @@ -95,7 +99,12 @@ func (p *Projection) PublishSessionStatus(status *compassv1.AgentSessionStatus) } p.mu.Lock() defer p.mu.Unlock() - p.sessions[status.GetSessionId()] = sessionEntry{state: status.GetState(), account: status.GetAgentAccountId()} + p.sessions[status.GetSessionId()] = sessionEntry{ + state: status.GetState(), + account: status.GetAgentAccountId(), + tier: status.GetRuntimeTier(), + egressPosture: status.GetEgressPosture(), + } p.bus.Publish(&compassv1.SubscribeEventsResponse{ Payload: &compassv1.SubscribeEventsResponse_AgentSessionStatus{ @@ -151,7 +160,14 @@ func isTerminal(state compassv1.AgentSessionState) bool { } // statusOf builds one board entry from a retained session record, carrying the -// DL-167 agent_account_id alongside the state. +// DL-167 agent_account_id and the owning Runner's runtime tier and egress +// posture alongside the state. func statusOf(sessionID string, entry sessionEntry) *compassv1.AgentSessionStatus { - return &compassv1.AgentSessionStatus{SessionId: sessionID, State: entry.state, AgentAccountId: entry.account} + return &compassv1.AgentSessionStatus{ + SessionId: sessionID, + State: entry.state, + AgentAccountId: entry.account, + RuntimeTier: entry.tier, + EgressPosture: entry.egressPosture, + } } diff --git a/go/internal/gen/compass/v1/runner.pb.go b/go/internal/gen/compass/v1/runner.pb.go index 70430c697..c2a630358 100644 --- a/go/internal/gen/compass/v1/runner.pb.go +++ b/go/internal/gen/compass/v1/runner.pb.go @@ -129,9 +129,16 @@ func (RunnerErrorCode) EnumDescriptor() ([]byte, []int) { // CodeUnauthenticated, so the field is a defense-in-depth cross-check, not a // trusted input. The credential itself rides the transport as a bearer token, // never a field here (mirroring IssueTokenRequest, compass.proto:237-242). +// +// runtime_tier and egress_posture are declared once here, not per session: +// they are Runner-wide facts of the one backend this Runner drives, so the hub +// stamps them onto every session it owns rather than have each lifecycle frame +// repeat them. type EnrollRequest struct { state protoimpl.MessageState `protogen:"open.v1"` RunnerId string `protobuf:"bytes,1,opt,name=runner_id,json=runnerId,proto3" json:"runner_id,omitempty"` + RuntimeTier v1.RuntimeTier `protobuf:"varint,2,opt,name=runtime_tier,json=runtimeTier,proto3,enum=compass.v1.RuntimeTier" json:"runtime_tier,omitempty"` + EgressPosture v1.EgressPosture `protobuf:"varint,3,opt,name=egress_posture,json=egressPosture,proto3,enum=compass.v1.EgressPosture" json:"egress_posture,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -173,6 +180,20 @@ func (x *EnrollRequest) GetRunnerId() string { return "" } +func (x *EnrollRequest) GetRuntimeTier() v1.RuntimeTier { + if x != nil { + return x.RuntimeTier + } + return v1.RuntimeTier(0) +} + +func (x *EnrollRequest) GetEgressPosture() v1.EgressPosture { + if x != nil { + return x.EgressPosture + } + return v1.EgressPosture(0) +} + // Enroll response: the handshake ack. `reattached` distinguishes a fresh // enrollment from a re-attach of an already-registered Runner (OQ6 duplicate // enrollment) — the wire-level evidence a @@ -2016,9 +2037,11 @@ var File_compass_v1_runner_proto protoreflect.FileDescriptor const file_compass_v1_runner_proto_rawDesc = "" + "\n" + "\x17compass/v1/runner.proto\x12\n" + - "compass.v1\x1a\x1ecompass/v1/agent_gateway.proto\x1a\x16compass/v1/agent.proto\x1a\x18compass/v1/compass.proto\x1a\x16compass/v1/forge.proto\",\n" + + "compass.v1\x1a\x1ecompass/v1/agent_gateway.proto\x1a\x16compass/v1/agent.proto\x1a\x18compass/v1/compass.proto\x1a\x16compass/v1/forge.proto\"\xaa\x01\n" + "\rEnrollRequest\x12\x1b\n" + - "\trunner_id\x18\x01 \x01(\tR\brunnerId\"0\n" + + "\trunner_id\x18\x01 \x01(\tR\brunnerId\x12:\n" + + "\fruntime_tier\x18\x02 \x01(\x0e2\x17.compass.v1.RuntimeTierR\vruntimeTier\x12@\n" + + "\x0eegress_posture\x18\x03 \x01(\x0e2\x19.compass.v1.EgressPostureR\regressPosture\"0\n" + "\x0eEnrollResponse\x12\x1e\n" + "\n" + "reattached\x18\x01 \x01(\bR\n" + @@ -2191,91 +2214,95 @@ var file_compass_v1_runner_proto_goTypes = []any{ (*RelayBoardCallResponse)(nil), // 24: compass.v1.RelayBoardCallResponse (*CommitConversationFrameRequest)(nil), // 25: compass.v1.CommitConversationFrameRequest (*CommitConversationFrameResponse)(nil), // 26: compass.v1.CommitConversationFrameResponse - (*v1.StartAgentSessionResponse)(nil), // 27: compass.v1.StartAgentSessionResponse - (*v1.StopAgentSessionResponse)(nil), // 28: compass.v1.StopAgentSessionResponse - (*v1.ReloadAgentSessionResponse)(nil), // 29: compass.v1.ReloadAgentSessionResponse - (*v1.GetAgentStatusResponse)(nil), // 30: compass.v1.GetAgentStatusResponse - (*v1.ProvisionAgentWorkspaceResponse)(nil), // 31: compass.v1.ProvisionAgentWorkspaceResponse - (*v1.RemoveAgentWorkspaceResponse)(nil), // 32: compass.v1.RemoveAgentWorkspaceResponse - (*v1.StartAgentSessionRequest)(nil), // 33: compass.v1.StartAgentSessionRequest - (*v1.StopAgentSessionRequest)(nil), // 34: compass.v1.StopAgentSessionRequest - (*v1.ReloadAgentSessionRequest)(nil), // 35: compass.v1.ReloadAgentSessionRequest - (*v1.GetAgentStatusRequest)(nil), // 36: compass.v1.GetAgentStatusRequest - (*v1.ProvisionAgentWorkspaceRequest)(nil), // 37: compass.v1.ProvisionAgentWorkspaceRequest - (*ForgeNotification)(nil), // 38: compass.v1.ForgeNotification - (*v1.RemoveAgentWorkspaceRequest)(nil), // 39: compass.v1.RemoveAgentWorkspaceRequest - (*AgentControl)(nil), // 40: compass.v1.AgentControl - (v1.SecretDelivery)(0), // 41: compass.v1.SecretDelivery - (v1.SecretKind)(0), // 42: compass.v1.SecretKind - (*AgentFrame)(nil), // 43: compass.v1.AgentFrame - (*CommsCallRequest)(nil), // 44: compass.v1.CommsCallRequest - (*CommsCallResult)(nil), // 45: compass.v1.CommsCallResult - (*LifecycleCallRequest)(nil), // 46: compass.v1.LifecycleCallRequest - (*LifecycleCallResult)(nil), // 47: compass.v1.LifecycleCallResult - (*ForgeCallRequest)(nil), // 48: compass.v1.ForgeCallRequest - (*ForgeCallResult)(nil), // 49: compass.v1.ForgeCallResult - (*BoardCallRequest)(nil), // 50: compass.v1.BoardCallRequest - (*BoardCallResult)(nil), // 51: compass.v1.BoardCallResult + (v1.RuntimeTier)(0), // 27: compass.v1.RuntimeTier + (v1.EgressPosture)(0), // 28: compass.v1.EgressPosture + (*v1.StartAgentSessionResponse)(nil), // 29: compass.v1.StartAgentSessionResponse + (*v1.StopAgentSessionResponse)(nil), // 30: compass.v1.StopAgentSessionResponse + (*v1.ReloadAgentSessionResponse)(nil), // 31: compass.v1.ReloadAgentSessionResponse + (*v1.GetAgentStatusResponse)(nil), // 32: compass.v1.GetAgentStatusResponse + (*v1.ProvisionAgentWorkspaceResponse)(nil), // 33: compass.v1.ProvisionAgentWorkspaceResponse + (*v1.RemoveAgentWorkspaceResponse)(nil), // 34: compass.v1.RemoveAgentWorkspaceResponse + (*v1.StartAgentSessionRequest)(nil), // 35: compass.v1.StartAgentSessionRequest + (*v1.StopAgentSessionRequest)(nil), // 36: compass.v1.StopAgentSessionRequest + (*v1.ReloadAgentSessionRequest)(nil), // 37: compass.v1.ReloadAgentSessionRequest + (*v1.GetAgentStatusRequest)(nil), // 38: compass.v1.GetAgentStatusRequest + (*v1.ProvisionAgentWorkspaceRequest)(nil), // 39: compass.v1.ProvisionAgentWorkspaceRequest + (*ForgeNotification)(nil), // 40: compass.v1.ForgeNotification + (*v1.RemoveAgentWorkspaceRequest)(nil), // 41: compass.v1.RemoveAgentWorkspaceRequest + (*AgentControl)(nil), // 42: compass.v1.AgentControl + (v1.SecretDelivery)(0), // 43: compass.v1.SecretDelivery + (v1.SecretKind)(0), // 44: compass.v1.SecretKind + (*AgentFrame)(nil), // 45: compass.v1.AgentFrame + (*CommsCallRequest)(nil), // 46: compass.v1.CommsCallRequest + (*CommsCallResult)(nil), // 47: compass.v1.CommsCallResult + (*LifecycleCallRequest)(nil), // 48: compass.v1.LifecycleCallRequest + (*LifecycleCallResult)(nil), // 49: compass.v1.LifecycleCallResult + (*ForgeCallRequest)(nil), // 50: compass.v1.ForgeCallRequest + (*ForgeCallResult)(nil), // 51: compass.v1.ForgeCallResult + (*BoardCallRequest)(nil), // 52: compass.v1.BoardCallRequest + (*BoardCallResult)(nil), // 53: compass.v1.BoardCallResult } var file_compass_v1_runner_proto_depIdxs = []int32{ - 27, // 0: compass.v1.SessionsRequest.start:type_name -> compass.v1.StartAgentSessionResponse - 28, // 1: compass.v1.SessionsRequest.stop:type_name -> compass.v1.StopAgentSessionResponse - 29, // 2: compass.v1.SessionsRequest.reload:type_name -> compass.v1.ReloadAgentSessionResponse - 30, // 3: compass.v1.SessionsRequest.status:type_name -> compass.v1.GetAgentStatusResponse - 31, // 4: compass.v1.SessionsRequest.provision:type_name -> compass.v1.ProvisionAgentWorkspaceResponse - 14, // 5: compass.v1.SessionsRequest.error:type_name -> compass.v1.RunnerError - 32, // 6: compass.v1.SessionsRequest.remove:type_name -> compass.v1.RemoveAgentWorkspaceResponse - 33, // 7: compass.v1.SessionsResponse.start:type_name -> compass.v1.StartAgentSessionRequest - 34, // 8: compass.v1.SessionsResponse.stop:type_name -> compass.v1.StopAgentSessionRequest - 35, // 9: compass.v1.SessionsResponse.reload:type_name -> compass.v1.ReloadAgentSessionRequest - 36, // 10: compass.v1.SessionsResponse.status:type_name -> compass.v1.GetAgentStatusRequest - 37, // 11: compass.v1.SessionsResponse.provision:type_name -> compass.v1.ProvisionAgentWorkspaceRequest - 38, // 12: compass.v1.SessionsResponse.forge_notification:type_name -> compass.v1.ForgeNotification - 10, // 13: compass.v1.SessionsResponse.secrets_version:type_name -> compass.v1.SecretsVersion - 13, // 14: compass.v1.SessionsResponse.config_version:type_name -> compass.v1.ConfigVersion - 6, // 15: compass.v1.SessionsResponse.deliver_control:type_name -> compass.v1.DispatchControl - 39, // 16: compass.v1.SessionsResponse.remove:type_name -> compass.v1.RemoveAgentWorkspaceRequest - 5, // 17: compass.v1.SessionsResponse.resume_body:type_name -> compass.v1.ResumeBody - 40, // 18: compass.v1.DispatchControl.op:type_name -> compass.v1.AgentControl - 9, // 19: compass.v1.FetchSecretsResponse.secrets:type_name -> compass.v1.ResolvedSecret - 41, // 20: compass.v1.ResolvedSecret.delivery:type_name -> compass.v1.SecretDelivery - 42, // 21: compass.v1.ResolvedSecret.kind:type_name -> compass.v1.SecretKind - 0, // 22: compass.v1.RunnerError.code:type_name -> compass.v1.RunnerErrorCode - 43, // 23: compass.v1.PublishEventsRequest.frame:type_name -> compass.v1.AgentFrame - 44, // 24: compass.v1.RelayCommsCallRequest.call:type_name -> compass.v1.CommsCallRequest - 45, // 25: compass.v1.RelayCommsCallResponse.result:type_name -> compass.v1.CommsCallResult - 46, // 26: compass.v1.RelayLifecycleCallRequest.call:type_name -> compass.v1.LifecycleCallRequest - 47, // 27: compass.v1.RelayLifecycleCallResponse.result:type_name -> compass.v1.LifecycleCallResult - 48, // 28: compass.v1.RelayForgeCallRequest.call:type_name -> compass.v1.ForgeCallRequest - 49, // 29: compass.v1.RelayForgeCallResponse.result:type_name -> compass.v1.ForgeCallResult - 50, // 30: compass.v1.RelayBoardCallRequest.call:type_name -> compass.v1.BoardCallRequest - 51, // 31: compass.v1.RelayBoardCallResponse.result:type_name -> compass.v1.BoardCallResult - 43, // 32: compass.v1.CommitConversationFrameRequest.frame:type_name -> compass.v1.AgentFrame - 1, // 33: compass.v1.RunnerService.Enroll:input_type -> compass.v1.EnrollRequest - 3, // 34: compass.v1.RunnerService.Sessions:input_type -> compass.v1.SessionsRequest - 15, // 35: compass.v1.RunnerService.PublishEvents:input_type -> compass.v1.PublishEventsRequest - 17, // 36: compass.v1.RunnerService.RelayCommsCall:input_type -> compass.v1.RelayCommsCallRequest - 19, // 37: compass.v1.RunnerService.RelayLifecycleCall:input_type -> compass.v1.RelayLifecycleCallRequest - 21, // 38: compass.v1.RunnerService.RelayForgeCall:input_type -> compass.v1.RelayForgeCallRequest - 23, // 39: compass.v1.RunnerService.RelayBoardCall:input_type -> compass.v1.RelayBoardCallRequest - 25, // 40: compass.v1.RunnerService.CommitConversationFrame:input_type -> compass.v1.CommitConversationFrameRequest - 7, // 41: compass.v1.RunnerService.FetchSecrets:input_type -> compass.v1.FetchSecretsRequest - 11, // 42: compass.v1.RunnerService.FetchAgentConfig:input_type -> compass.v1.FetchAgentConfigRequest - 2, // 43: compass.v1.RunnerService.Enroll:output_type -> compass.v1.EnrollResponse - 4, // 44: compass.v1.RunnerService.Sessions:output_type -> compass.v1.SessionsResponse - 16, // 45: compass.v1.RunnerService.PublishEvents:output_type -> compass.v1.PublishEventsResponse - 18, // 46: compass.v1.RunnerService.RelayCommsCall:output_type -> compass.v1.RelayCommsCallResponse - 20, // 47: compass.v1.RunnerService.RelayLifecycleCall:output_type -> compass.v1.RelayLifecycleCallResponse - 22, // 48: compass.v1.RunnerService.RelayForgeCall:output_type -> compass.v1.RelayForgeCallResponse - 24, // 49: compass.v1.RunnerService.RelayBoardCall:output_type -> compass.v1.RelayBoardCallResponse - 26, // 50: compass.v1.RunnerService.CommitConversationFrame:output_type -> compass.v1.CommitConversationFrameResponse - 8, // 51: compass.v1.RunnerService.FetchSecrets:output_type -> compass.v1.FetchSecretsResponse - 12, // 52: compass.v1.RunnerService.FetchAgentConfig:output_type -> compass.v1.FetchAgentConfigResponse - 43, // [43:53] is the sub-list for method output_type - 33, // [33:43] is the sub-list for method input_type - 33, // [33:33] is the sub-list for extension type_name - 33, // [33:33] is the sub-list for extension extendee - 0, // [0:33] is the sub-list for field type_name + 27, // 0: compass.v1.EnrollRequest.runtime_tier:type_name -> compass.v1.RuntimeTier + 28, // 1: compass.v1.EnrollRequest.egress_posture:type_name -> compass.v1.EgressPosture + 29, // 2: compass.v1.SessionsRequest.start:type_name -> compass.v1.StartAgentSessionResponse + 30, // 3: compass.v1.SessionsRequest.stop:type_name -> compass.v1.StopAgentSessionResponse + 31, // 4: compass.v1.SessionsRequest.reload:type_name -> compass.v1.ReloadAgentSessionResponse + 32, // 5: compass.v1.SessionsRequest.status:type_name -> compass.v1.GetAgentStatusResponse + 33, // 6: compass.v1.SessionsRequest.provision:type_name -> compass.v1.ProvisionAgentWorkspaceResponse + 14, // 7: compass.v1.SessionsRequest.error:type_name -> compass.v1.RunnerError + 34, // 8: compass.v1.SessionsRequest.remove:type_name -> compass.v1.RemoveAgentWorkspaceResponse + 35, // 9: compass.v1.SessionsResponse.start:type_name -> compass.v1.StartAgentSessionRequest + 36, // 10: compass.v1.SessionsResponse.stop:type_name -> compass.v1.StopAgentSessionRequest + 37, // 11: compass.v1.SessionsResponse.reload:type_name -> compass.v1.ReloadAgentSessionRequest + 38, // 12: compass.v1.SessionsResponse.status:type_name -> compass.v1.GetAgentStatusRequest + 39, // 13: compass.v1.SessionsResponse.provision:type_name -> compass.v1.ProvisionAgentWorkspaceRequest + 40, // 14: compass.v1.SessionsResponse.forge_notification:type_name -> compass.v1.ForgeNotification + 10, // 15: compass.v1.SessionsResponse.secrets_version:type_name -> compass.v1.SecretsVersion + 13, // 16: compass.v1.SessionsResponse.config_version:type_name -> compass.v1.ConfigVersion + 6, // 17: compass.v1.SessionsResponse.deliver_control:type_name -> compass.v1.DispatchControl + 41, // 18: compass.v1.SessionsResponse.remove:type_name -> compass.v1.RemoveAgentWorkspaceRequest + 5, // 19: compass.v1.SessionsResponse.resume_body:type_name -> compass.v1.ResumeBody + 42, // 20: compass.v1.DispatchControl.op:type_name -> compass.v1.AgentControl + 9, // 21: compass.v1.FetchSecretsResponse.secrets:type_name -> compass.v1.ResolvedSecret + 43, // 22: compass.v1.ResolvedSecret.delivery:type_name -> compass.v1.SecretDelivery + 44, // 23: compass.v1.ResolvedSecret.kind:type_name -> compass.v1.SecretKind + 0, // 24: compass.v1.RunnerError.code:type_name -> compass.v1.RunnerErrorCode + 45, // 25: compass.v1.PublishEventsRequest.frame:type_name -> compass.v1.AgentFrame + 46, // 26: compass.v1.RelayCommsCallRequest.call:type_name -> compass.v1.CommsCallRequest + 47, // 27: compass.v1.RelayCommsCallResponse.result:type_name -> compass.v1.CommsCallResult + 48, // 28: compass.v1.RelayLifecycleCallRequest.call:type_name -> compass.v1.LifecycleCallRequest + 49, // 29: compass.v1.RelayLifecycleCallResponse.result:type_name -> compass.v1.LifecycleCallResult + 50, // 30: compass.v1.RelayForgeCallRequest.call:type_name -> compass.v1.ForgeCallRequest + 51, // 31: compass.v1.RelayForgeCallResponse.result:type_name -> compass.v1.ForgeCallResult + 52, // 32: compass.v1.RelayBoardCallRequest.call:type_name -> compass.v1.BoardCallRequest + 53, // 33: compass.v1.RelayBoardCallResponse.result:type_name -> compass.v1.BoardCallResult + 45, // 34: compass.v1.CommitConversationFrameRequest.frame:type_name -> compass.v1.AgentFrame + 1, // 35: compass.v1.RunnerService.Enroll:input_type -> compass.v1.EnrollRequest + 3, // 36: compass.v1.RunnerService.Sessions:input_type -> compass.v1.SessionsRequest + 15, // 37: compass.v1.RunnerService.PublishEvents:input_type -> compass.v1.PublishEventsRequest + 17, // 38: compass.v1.RunnerService.RelayCommsCall:input_type -> compass.v1.RelayCommsCallRequest + 19, // 39: compass.v1.RunnerService.RelayLifecycleCall:input_type -> compass.v1.RelayLifecycleCallRequest + 21, // 40: compass.v1.RunnerService.RelayForgeCall:input_type -> compass.v1.RelayForgeCallRequest + 23, // 41: compass.v1.RunnerService.RelayBoardCall:input_type -> compass.v1.RelayBoardCallRequest + 25, // 42: compass.v1.RunnerService.CommitConversationFrame:input_type -> compass.v1.CommitConversationFrameRequest + 7, // 43: compass.v1.RunnerService.FetchSecrets:input_type -> compass.v1.FetchSecretsRequest + 11, // 44: compass.v1.RunnerService.FetchAgentConfig:input_type -> compass.v1.FetchAgentConfigRequest + 2, // 45: compass.v1.RunnerService.Enroll:output_type -> compass.v1.EnrollResponse + 4, // 46: compass.v1.RunnerService.Sessions:output_type -> compass.v1.SessionsResponse + 16, // 47: compass.v1.RunnerService.PublishEvents:output_type -> compass.v1.PublishEventsResponse + 18, // 48: compass.v1.RunnerService.RelayCommsCall:output_type -> compass.v1.RelayCommsCallResponse + 20, // 49: compass.v1.RunnerService.RelayLifecycleCall:output_type -> compass.v1.RelayLifecycleCallResponse + 22, // 50: compass.v1.RunnerService.RelayForgeCall:output_type -> compass.v1.RelayForgeCallResponse + 24, // 51: compass.v1.RunnerService.RelayBoardCall:output_type -> compass.v1.RelayBoardCallResponse + 26, // 52: compass.v1.RunnerService.CommitConversationFrame:output_type -> compass.v1.CommitConversationFrameResponse + 8, // 53: compass.v1.RunnerService.FetchSecrets:output_type -> compass.v1.FetchSecretsResponse + 12, // 54: compass.v1.RunnerService.FetchAgentConfig:output_type -> compass.v1.FetchAgentConfigResponse + 45, // [45:55] is the sub-list for method output_type + 35, // [35:45] is the sub-list for method input_type + 35, // [35:35] is the sub-list for extension type_name + 35, // [35:35] is the sub-list for extension extendee + 0, // [0:35] is the sub-list for field type_name } func init() { file_compass_v1_runner_proto_init() } diff --git a/go/internal/runner/enroll_identity_test.go b/go/internal/runner/enroll_identity_test.go new file mode 100644 index 000000000..8ab31b4c8 --- /dev/null +++ b/go/internal/runner/enroll_identity_test.go @@ -0,0 +1,92 @@ +//go:build unix + +package runner + +// Dial declares the Runner's runtime tier and egress posture ONCE at enrollment, +// derived from the engine it drives. This drives the real Dial (interceptor- +// wrapped client) against a recording RunnerService handler and asserts the +// EnrollRequest carried the right wire enums for a HOST engine (tier HOST, egress +// UNENFORCED) — the two facts the hub stamps onto every session status. + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "connectrpc.com/connect" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/gen/compass/v1/compassv1internalconnect" + "github.com/RigelBuild/compass/go/internal/runtime" +) + +// recordingEnroll is a RunnerService handler that captures the EnrollRequest it +// received, so a Dial test can assert on the tier + posture the client declared. +type recordingEnroll struct { + compassv1internalconnect.UnimplementedRunnerServiceHandler + mu sync.Mutex + req *compassv1internal.EnrollRequest +} + +func (r *recordingEnroll) Enroll(_ context.Context, req *connect.Request[compassv1internal.EnrollRequest]) (*connect.Response[compassv1internal.EnrollResponse], error) { + r.mu.Lock() + defer r.mu.Unlock() + r.req = req.Msg + return connect.NewResponse(&compassv1internal.EnrollResponse{Reattached: false}), nil +} + +func (r *recordingEnroll) enrolled() *compassv1internal.EnrollRequest { + r.mu.Lock() + defer r.mu.Unlock() + return r.req +} + +// recordingEnrollServer stands up an h2c httptest RunnerService serving rec and +// returns its base URL, torn down via t.Cleanup. +func recordingEnrollServer(t *testing.T, rec *recordingEnroll) string { + t.Helper() + path, handler := compassv1internalconnect.NewRunnerServiceHandler(rec) + mux := http.NewServeMux() + mux.Handle(path, handler) + srv := httptest.NewUnstartedServer(mux) + srv.Config.Protocols = cleartextHTTP2() + srv.Start() + t.Cleanup(srv.Close) + return srv.URL +} + +// TestDialDeclaresEngineTierAndPosture pins the enrollment declaration: dialing +// with a HOST engine sends runtime_tier=HOST and egress_posture=UNENFORCED on the +// EnrollRequest, derived from the engine via runtime.TierOf / runtime.PostureOf. +// +// Negative control: dropping the two fields from Dial's EnrollRequest (the +// pre-fix shape, runner_id only) reddens both assertions — observed "runtime_tier +// = RUNTIME_TIER_UNSPECIFIED, want RUNTIME_TIER_HOST". +func TestDialDeclaresEngineTierAndPosture(t *testing.T) { + rec := &recordingEnroll{} + url := recordingEnrollServer(t, rec) + + // context.Background() is the test root context. + if _, err := Dial(context.Background(), RunnerConfig{ + RunnerID: "r-1", + ServerAddr: url, + Token: "tok", + Engine: runtime.NewHostRuntime(t.TempDir()), + }); err != nil { + t.Fatalf("Dial = %v, want success", err) + } + + got := rec.enrolled() + if got == nil { + t.Fatal("handler recorded no EnrollRequest") + } + if got.GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_HOST { + t.Errorf("EnrollRequest runtime_tier = %v, want RUNTIME_TIER_HOST", got.GetRuntimeTier()) + } + if got.GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED { + t.Errorf("EnrollRequest egress_posture = %v, want EGRESS_POSTURE_UNENFORCED", got.GetEgressPosture()) + } +} diff --git a/go/internal/runner/runner.go b/go/internal/runner/runner.go index 3e011a408..6e6f2bc0c 100644 --- a/go/internal/runner/runner.go +++ b/go/internal/runner/runner.go @@ -115,7 +115,9 @@ func Dial(ctx context.Context, cfg RunnerConfig) (*ServerLink, error) { connect.WithInterceptors(otelInterceptor, &bearerToken{token: cfg.Token}), ) resp, err := client.Enroll(ctx, connect.NewRequest(&compassv1internal.EnrollRequest{ - RunnerId: cfg.RunnerID, + RunnerId: cfg.RunnerID, + RuntimeTier: runtimeTierProto(runtime.TierOf(cfg.Engine)), + EgressPosture: egressPostureProto(runtime.PostureOf(cfg.Engine)), })) if err != nil { return nil, fmt.Errorf("enrolling runner %q: %w", cfg.RunnerID, err) diff --git a/go/internal/runnerhub/binding_cache_test.go b/go/internal/runnerhub/binding_cache_test.go index 40279e077..58d851e2c 100644 --- a/go/internal/runnerhub/binding_cache_test.go +++ b/go/internal/runnerhub/binding_cache_test.go @@ -31,6 +31,7 @@ import ( "sync" "testing" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/fabric" "github.com/RigelBuild/compass/go/internal/store" ) @@ -228,7 +229,7 @@ func TestRestartResolvesPreRestartBindingBothDirections(t *testing.T) { // A FIRST enroll on a fresh hub (the restart case): reattached is false, so // enroll does NOT reap the durable rows — they are still valid. - if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject()); reattached { + if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); reattached { t.Fatal("first enroll on a fresh hub reported reattached=true, want false (no durable reap)") } @@ -255,7 +256,7 @@ func TestFailClosedStoppedNeverSeenAndPostReconnect(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // Never-seen: no binding anywhere. if acct, ok := hub.accountForSession(context.Background(), "ghost"); ok { @@ -283,7 +284,7 @@ func TestFailClosedStoppedNeverSeenAndPostReconnect(t *testing.T) { t.Fatalf("accountForSession(sess-pre) before reconnect = (%q, %v), want (%s, true)", acct, ok, testAgentAccount) } // Runner reconnects: a re-enroll (reattached=true) durably reaps. - if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject()); !reattached { + if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); !reattached { t.Fatal("second enroll reported reattached=false, want true (a Runner reconnect)") } if acct, ok := hub.accountForSession(context.Background(), "sess-pre"); ok { @@ -320,7 +321,7 @@ func TestPeerBindingChangeEvictsOtherInstanceCache(t *testing.T) { bindingsB.seed("sess-old") hubB.SetSessionBindingStore(bindingsB) hubB.SetRoutingFabric(routing) - hubB.enroll(context.Background(), "runner-1", runnerSubject()) + hubB.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hubB.bindContainer("cont-new", testAgentAccount) // B re-points the account onto sess-new: displaces sess-old, publishes the @@ -360,7 +361,7 @@ func TestDisplacedSessionResolvesNowhere(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // Bind the account to sess-old. hub.bindContainer("cont-old", testAgentAccount) @@ -406,7 +407,7 @@ func TestAckPathBindingReadNeverRunsUnderSystemRole(t *testing.T) { hub.SetDeliveryStore(del) // Enrolled Runner + empty maps (no bindSession) => the ack's account // resolve is a cache MISS that falls through to the read-through table. - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if err := hub.Deliver(context.Background(), RunnerEvent{ RunnerSeq: 1, SessionID: "sess-1", Frame: deliveryAckFrame("m1"), @@ -434,7 +435,7 @@ func TestAckPathBindingReadNeverRunsUnderSystemRole(t *testing.T) { hub.SetSessionBindingStore(bindings) del := newFakeDeliveryStore() hub.SetDeliveryStore(del) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if err := hub.Deliver(context.Background(), RunnerEvent{ RunnerSeq: 1, SessionID: "sess-1", Frame: forgeAckFrame("sub-1"), @@ -470,7 +471,7 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { routing := &fakeRoutingFabric{} hub.SetSessionBindingStore(bindings) hub.SetRoutingFabric(routing) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) hub.promoteSession(context.Background(), "cont-1", "sess-1") @@ -488,13 +489,13 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) hub.promoteSession(context.Background(), "cont-1", "sess-1") // The reconnect sweep now faults; the in-RAM snapshot must still drive it. bindings.deleteForRunnerErr = faultErr - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if acct, ok := hub.accountForSession(context.Background(), "sess-1"); ok { t.Fatalf("accountForSession(sess-1) = (%q, true) after a reconnect, want fail-closed: a reap fault must not leave a dead session resolvable", acct) @@ -507,7 +508,7 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { bindings.seed("sess-1") bindings.resolveErr = faultErr hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if acct, ok := hub.accountForSession(context.Background(), "sess-1"); ok { t.Fatalf("accountForSession(sess-1) = (%q, true), want fail-closed: a store fault must never resolve", acct) @@ -520,7 +521,7 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { bindings.seed("sess-1") bindings.reverseErr = faultErr hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if sess, ok := hub.SessionForAccount(context.Background(), testAgentAccount); ok { t.Fatalf("SessionForAccount = (%q, true), want fail-closed: a store fault must never resolve", sess) @@ -541,7 +542,7 @@ func TestReusedSessionIDConflictIsSwallowed(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // A row from before the joint restart, under a DIFFERENT account. bindings.mu.Lock() @@ -580,7 +581,7 @@ func TestConcurrentResolveDuringAFaultingReapCannotResurrect(t *testing.T) { release: make(chan struct{}), } hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) hub.promoteSession(context.Background(), "cont-1", "sess-1") @@ -590,7 +591,7 @@ func TestConcurrentResolveDuringAFaultingReapCannotResurrect(t *testing.T) { done := make(chan struct{}) go func() { defer close(done) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) }() <-bindings.entered // maps are cleared; the reap is in flight and will fail diff --git a/go/internal/runnerhub/commands_test.go b/go/internal/runnerhub/commands_test.go index 5be44ff5b..fc90ae9ad 100644 --- a/go/internal/runnerhub/commands_test.go +++ b/go/internal/runnerhub/commands_test.go @@ -99,7 +99,7 @@ func TestStartRelaySurfacesAlreadyRunningAsAlreadyExists(t *testing.T) { hub := newHubOnly() // 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(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) 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) { // exercised. func TestStartRelayReturnsSessionIdOnSuccess(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") router.attach(func(cmd *compassv1internal.SessionsResponse) error { go router.complete(&compassv1internal.SessionsRequest{ @@ -156,7 +156,7 @@ func TestStartRelayReturnsSessionIdOnSuccess(t *testing.T) { // from. A non-zero count means the initial-signal path was re-introduced. func TestStartEmitsNoInitialSignal(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", testAgentAccount) router, _, _ := hub.routerFor("any") rec := newRecordingSend() @@ -187,7 +187,7 @@ func TestStartEmitsNoInitialSignal(t *testing.T) { // variant, and the typed result flows back. func TestRemoveRelayReturnsResponseOnSuccess(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") var sawRemove bool router.attach(func(cmd *compassv1internal.SessionsResponse) error { @@ -223,7 +223,7 @@ func TestRemoveRelayReturnsResponseOnSuccess(t *testing.T) { // true after teardown and reddens this. func TestRemoveClearsContainerBinding(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", testAgentAccount) if !hub.HasContainerBinding("c1") { t.Fatal("precondition: container c1 should be bound after bindContainer") @@ -250,7 +250,7 @@ func TestRemoveClearsContainerBinding(t *testing.T) { // the pushed request id — the seam the SessionState fallback tests drive. func attachStatusResponder(t *testing.T, hub *Hub, statuses []*compassv1.AgentSessionStatus) { t.Helper() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) diff --git a/go/internal/runnerhub/config_fetch_test.go b/go/internal/runnerhub/config_fetch_test.go index 191aae20f..7c1e39955 100644 --- a/go/internal/runnerhub/config_fetch_test.go +++ b/go/internal/runnerhub/config_fetch_test.go @@ -25,6 +25,7 @@ import ( "connectrpc.com/connect" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/store" ) @@ -83,7 +84,7 @@ func drainConfigStream(t *testing.T, stream *connect.ServerStreamForClient[compa // from the CodeUnavailable of a transient transport fault. func TestFetchAgentConfigNoConfigStoreFailsPrecondition(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, nil) client := newRawRunnerClient(t, url, "runner-tok") @@ -108,7 +109,7 @@ func TestFetchAgentConfigNoConfigStoreFailsPrecondition(t *testing.T) { // loop is exercised (a bug that sent one frame, or dropped the tail, reds). func TestFetchAgentConfigStreamsVersionThenChunks(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // A bundle spanning multiple chunk frames. want := bytes.Repeat([]byte("compass-config-"), configChunkBytes/10) cfg := &fakeConfigStore{version: "v-1", bundle: want} @@ -137,7 +138,7 @@ func TestFetchAgentConfigStreamsVersionThenChunks(t *testing.T) { // state (the Runner materializes an empty dir), never an error. func TestFetchAgentConfigUnconfiguredEmptyVersion(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{err: store.ErrNotFound} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") @@ -161,7 +162,7 @@ func TestFetchAgentConfigUnconfiguredEmptyVersion(t *testing.T) { // not read as "no config". func TestFetchAgentConfigStoreErrorMapsToInternal(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{err: errors.New("db boom")} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") @@ -186,7 +187,7 @@ func TestFetchAgentConfigStoreErrorMapsToInternal(t *testing.T) { // that streamed the bundle anyway (wasting the reconnect) reds. func TestFetchAgentConfigIfVersionMatchStreamsVersionOnly(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{version: "v-held", bundle: []byte("should-not-be-sent")} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") @@ -210,7 +211,7 @@ func TestFetchAgentConfigIfVersionMatchStreamsVersionOnly(t *testing.T) { // a stale Runner reconnecting with an old version still gets the new bytes. func TestFetchAgentConfigIfVersionMismatchStreamsBundle(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{version: "v-new", bundle: []byte("new-bytes")} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") diff --git a/go/internal/runnerhub/config_signal_test.go b/go/internal/runnerhub/config_signal_test.go index 5e5e860b8..1a432ddd9 100644 --- a/go/internal/runnerhub/config_signal_test.go +++ b/go/internal/runnerhub/config_signal_test.go @@ -14,6 +14,7 @@ import ( "context" "testing" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/store" ) @@ -25,7 +26,7 @@ import ( // version), never a minted token. func TestSignalConfigVersionPushesStoreVersion(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-a") bindSession(hub, "sess-b") router, _, err := hub.routerFor("any") @@ -60,7 +61,7 @@ func TestSignalConfigVersionPushesStoreVersion(t *testing.T) { // fleet-cleared marker the Runner reads as "materialize an empty dir". func TestSignalConfigVersionEmptyVersionIsTheClearedMarker(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-a") router, _, err := hub.routerFor("any") if err != nil { @@ -87,7 +88,7 @@ func TestSignalConfigVersionEmptyVersionIsTheClearedMarker(t *testing.T) { // (nothing bound) pushes nothing and is a clean success. func TestSignalConfigVersionNoLiveSessionsIsNoop(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") rec := newRecordingSend() router.attach(rec.send) diff --git a/go/internal/runnerhub/deliveryarm_test.go b/go/internal/runnerhub/deliveryarm_test.go index 66cb1be08..4b57ac0bf 100644 --- a/go/internal/runnerhub/deliveryarm_test.go +++ b/go/internal/runnerhub/deliveryarm_test.go @@ -349,7 +349,7 @@ func TestDeliverSessionNilPresenceSinkIsSafe(t *testing.T) { // DispatchControl returns without blocking. func TestDispatchControlSendOnlyDoesNotBlock(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("sess-1") if err != nil { t.Fatalf("routerFor: %v", err) @@ -400,7 +400,7 @@ func TestDispatchControlSendOnlyDoesNotBlock(t *testing.T) { // is observed (counted), not dropped as unknown. func TestDispatchControlNoLiveStreamRefuses(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // No stream attached: send is nil. op := &compassv1internal.AgentControl{ Control: &compassv1internal.AgentControl_Deliver{ diff --git a/go/internal/runnerhub/enroll_reap_test.go b/go/internal/runnerhub/enroll_reap_test.go index 18fd6c6dd..87617bb87 100644 --- a/go/internal/runnerhub/enroll_reap_test.go +++ b/go/internal/runnerhub/enroll_reap_test.go @@ -17,6 +17,7 @@ import ( "sync" "testing" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/store" ) @@ -53,7 +54,7 @@ func TestEnrollFiresReapSinkWithClearedSessionIDs(t *testing.T) { hub.SetSessionReapSink(fake) // A first enroll binds the Runner, then two live sessions promote onto it. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", "acct-a") hub.promoteSession(context.Background(), "c1", "sess-a") hub.bindContainer("c2", "acct-b") @@ -66,7 +67,7 @@ func TestEnrollFiresReapSinkWithClearedSessionIDs(t *testing.T) { } // The Runner reconnects: enroll clears both bindings and reaps both ids. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) calls := fake.snapshot() if len(calls) != 2 { @@ -85,12 +86,12 @@ func TestEnrollFiresReapSinkWithClearedSessionIDs(t *testing.T) { func TestEnrollNilReapSinkStillClears(t *testing.T) { hub := newHubOnly() // no SetSessionReapSink - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", "acct-a") hub.promoteSession(context.Background(), "c1", "sess-a") // A re-enroll with no reap sink clears the binding without panicking. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if sess, ok := hub.SessionForAccount(context.Background(), "acct-a"); ok { t.Fatalf("SessionForAccount(acct-a) = %q ok=true after re-enroll, want ok=false (binding cleared)", sess) diff --git a/go/internal/runnerhub/handler.go b/go/internal/runnerhub/handler.go index 943ec071a..2c26ed405 100644 --- a/go/internal/runnerhub/handler.go +++ b/go/internal/runnerhub/handler.go @@ -82,7 +82,7 @@ func (h *Handler) Enroll(ctx context.Context, req *connect.Request[compassv1inte // enroll under an identity other than its token's. return nil, errUnauthenticated } - reattached := h.hub.enroll(ctx, subj.ID, subj) + reattached := h.hub.enroll(ctx, subj.ID, subj, req.Msg.GetRuntimeTier(), req.Msg.GetEgressPosture()) return connect.NewResponse(&compassv1internal.EnrollResponse{Reattached: reattached}), nil } diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index 2432cdd4b..f6d978190 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -486,9 +486,11 @@ type Hub struct { // attachedRunner is one enrolled Runner: its id, its authenticated token // subject, and the command router that reaches its live Sessions stream. type attachedRunner struct { - id string - subject store.Subject - router *commandRouter + id string + subject store.Subject + router *commandRouter + tier compassv1.RuntimeTier + egressPosture compassv1.EgressPosture } // NewHub constructs a hub over the two write-through sinks and the agent-comms @@ -801,8 +803,13 @@ func (h *Hub) deliverSession(ctx context.Context, sessionID string, sf *compassv // unbindSession has not yet dropped it) carries its account; one published // after a Runner reconnect cleared the maps carries none (the stated residual // gap). accountForSession takes h.mu; deliverSession holds no lock here. + // runnerRuntimeIdentity reads the enrolled Runner's tier/posture under the + // same lock, so a status published after a reattach reflects the newly + // enrolled Runner's values. It is a separate critical section from the + // account resolve above, not one atomic read of both. account, hasAccount := h.accountForSession(ctx, sessionID) - status := &compassv1.AgentSessionStatus{SessionId: sessionID, State: state} + tier, egressPosture := h.runnerRuntimeIdentity() + status := &compassv1.AgentSessionStatus{SessionId: sessionID, State: state, RuntimeTier: tier, EgressPosture: egressPosture} if hasAccount { status.AgentAccountId = string(account) } @@ -1045,12 +1052,12 @@ type promotedPair struct { // // A hub with no binding store wired keeps the original in-RAM snapshot behaviour // (every existing enroll test), driving offline/reapedSessions from the maps. -func (h *Hub) enroll(ctx context.Context, id string, subject store.Subject) (reattached bool) { +func (h *Hub) enroll(ctx context.Context, id string, subject store.Subject, tier compassv1.RuntimeTier, egressPosture compassv1.EgressPosture) (reattached bool) { h.mu.Lock() reattached = h.runner != nil router := newCommandRouter() router.log = h.log - h.runner = &attachedRunner{id: id, subject: subject, router: router} + h.runner = &attachedRunner{id: id, subject: subject, router: router, tier: tier, egressPosture: egressPosture} // Snapshot the live (account -> session) bindings BEFORE clearing them, for // the no-store path: each previously-bound account loses its live session on // this re-enroll and must be driven to presence OFFLINE (RIG-1569 T8). enroll @@ -1155,3 +1162,18 @@ func (h *Hub) routerFor(sessionID string) (*commandRouter, string, error) { } return h.runner.router, h.runner.id, nil } + +// runnerRuntimeIdentity returns the enrolled Runner's declared runtime tier and +// egress posture under h.mu, so a session status stamps the Runner that owns it +// today rather than racing a re-enroll. No Runner enrolled yields UNSPECIFIED on +// both — the wire's "we do not know", never a plausible default. h.runner is +// assigned only in enroll and never set back to nil, so no disconnect path can +// regress a known tier to UNSPECIFIED. +func (h *Hub) runnerRuntimeIdentity() (compassv1.RuntimeTier, compassv1.EgressPosture) { + h.mu.Lock() + defer h.mu.Unlock() + if h.runner == nil { + return compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED + } + return h.runner.tier, h.runner.egressPosture +} diff --git a/go/internal/runnerhub/hub_test.go b/go/internal/runnerhub/hub_test.go index 48894092b..9683c3c39 100644 --- a/go/internal/runnerhub/hub_test.go +++ b/go/internal/runnerhub/hub_test.go @@ -167,10 +167,10 @@ func TestEnrollDuplicateReattaches(t *testing.T) { hub := newHubOnly() subj := store.Subject{Kind: store.SubjectRunner, ID: "runner-1"} - if reattached := hub.enroll(context.Background(), "runner-1", subj); reattached { + if reattached := hub.enroll(context.Background(), "runner-1", subj, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); reattached { t.Fatal("first enroll reattached = true, want false (fresh registration)") } - if reattached := hub.enroll(context.Background(), "runner-1", subj); !reattached { + if reattached := hub.enroll(context.Background(), "runner-1", subj, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); !reattached { t.Fatal("second enroll reattached = false, want true (single-Runner MVP re-attaches)") } // A router is resolvable after enrollment (a session command has a Runner to diff --git a/go/internal/runnerhub/provision_dedup_test.go b/go/internal/runnerhub/provision_dedup_test.go index 46f6b90cd..819cf2e53 100644 --- a/go/internal/runnerhub/provision_dedup_test.go +++ b/go/internal/runnerhub/provision_dedup_test.go @@ -52,7 +52,7 @@ type provisionOutcome struct { // Hub.Provision through the real router. func enrollAttached(t *testing.T, hub *Hub, send *recordingSend) *commandRouter { t.Helper() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want the live router", err) diff --git a/go/internal/runnerhub/relay_comms_test.go b/go/internal/runnerhub/relay_comms_test.go index 222163700..540c2062d 100644 --- a/go/internal/runnerhub/relay_comms_test.go +++ b/go/internal/runnerhub/relay_comms_test.go @@ -270,7 +270,7 @@ func TestRelayCommsCallDropsBindingOnRunnerReconnect(t *testing.T) { } // The Runner reconnects (re-enroll), which drops ALL agent-comms bindings. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // The SAME session_id now fails closed — the binding is gone, so no stale // account is reachable. @@ -330,7 +330,7 @@ func TestRelayCommsCallStoppedSessionFailsClosedNotFound(t *testing.T) { // helper. func TestProvisionThenStartBindsSessionToProvisionedAccount(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) @@ -377,7 +377,7 @@ func TestProvisionThenStartBindsSessionToProvisionedAccount(t *testing.T) { // RelayCommsCall fails closed CodeNotFound — never an empty-account attribution. func TestProvisionWithEmptyAccountLeavesNoBindingAndFailsClosed(t *testing.T) { hub, comms := newHubWithComms() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) @@ -439,7 +439,7 @@ func TestProvisionWithEmptyAccountLeavesNoBindingAndFailsClosed(t *testing.T) { // re-enrolls, and asserts the reverse map is empty. func TestEnrollClearsReverseAccountSessions(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindLiveSession(hub) // acct-agent -> sess-1, via the real Provision->Start path // Sanity: the reverse map is populated before the re-enroll. @@ -449,7 +449,7 @@ func TestEnrollClearsReverseAccountSessions(t *testing.T) { // A Runner reconnect: enroll re-attaches and MUST drop every stale binding, // forward AND reverse. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if sess, ok := hub.SessionForAccount(context.Background(), "acct-agent"); ok { t.Fatalf("SessionForAccount(acct-agent) = %q, ok=true after re-enroll; want ok=false — enroll left a stale reverse entry, so a dead session resolves as live", sess) @@ -523,7 +523,7 @@ func TestEnrollFiresTerminalPresenceEdgePerBoundAccountAndClears(t *testing.T) { hub := newHubOnly() pres := &fakePresenceSink{} hub.SetPresenceSink(pres) - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", "acct-a") hub.promoteSession(context.Background(), "c1", "sess-a") hub.bindContainer("c2", "acct-b") @@ -531,7 +531,7 @@ func TestEnrollFiresTerminalPresenceEdgePerBoundAccountAndClears(t *testing.T) { // A Runner reconnect: enroll drops every binding and drives each previously- // bound account OFFLINE. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) life := pres.lifecycleSnapshot() if len(life) != 2 { @@ -563,7 +563,7 @@ func TestFirstEnrollFiresNoTerminalPresenceEdge(t *testing.T) { pres := &fakePresenceSink{} hub.SetPresenceSink(pres) - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if life := pres.lifecycleSnapshot(); len(life) != 0 { t.Fatalf("lifecycle edges after first enroll = %d, want 0 (nothing was bound): %+v", len(life), life) diff --git a/go/internal/runnerhub/relay_operator_fault_test.go b/go/internal/runnerhub/relay_operator_fault_test.go index 733793c4f..51d661bfa 100644 --- a/go/internal/runnerhub/relay_operator_fault_test.go +++ b/go/internal/runnerhub/relay_operator_fault_test.go @@ -24,7 +24,7 @@ import ( // See docs/designs/infra/runtime/compass-runner-gateway-error-sentinels/design.md. func TestProvisionRelaySurfacesOperatorFaultAsFailedPrecondition(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) @@ -34,7 +34,7 @@ func TestProvisionRelaySurfacesOperatorFaultAsFailedPrecondition(t *testing.T) { // the socket diagnostic followed by the appended gateway.ErrOperatorConfig // sentinel text. The Runner is simulated here, so the wire carries only the // string — the gateway package is deliberately not imported. - const diag = "serving agent socket for container \"cont-op\": agent socket path \"/run/compass/containers/cont-op/agent.sock\" is 120 bytes, over the 108-byte AF_UNIX limit: shorten the socket's parent directory or the agent account id: operator-fault runner configuration" + const diag = "serving agent socket for container \"cont-op\": agent socket path \"/run/compass/containers/cont-op/agent.sock\" is 120 bytes, over the 108-byte AF_UNIX limit: shorten the Runner's --runtime-dir or the agent account id: operator-fault runner configuration" router.attach(func(cmd *compassv1internal.SessionsResponse) error { go router.complete(&compassv1internal.SessionsRequest{ diff --git a/go/internal/runnerhub/runtime_identity_test.go b/go/internal/runnerhub/runtime_identity_test.go new file mode 100644 index 000000000..72d770a73 --- /dev/null +++ b/go/internal/runnerhub/runtime_identity_test.go @@ -0,0 +1,181 @@ +//go:build unix + +package runnerhub + +// The enrollment-carried runtime tier + egress posture reach BOTH production +// render paths. A Runner declares its tier and posture ONCE at enrollment; the +// hub stamps them onto every session status it publishes. These tests drive a +// session lifecycle frame through the REAL hub wired to the REAL Bridge board +// (board.Projection as the LifecycleSink), so one delivery exercises the two +// surfaces off one source of truth: +// - the UI's SubscribeEvents fan-out (bus.Subscribe -> Live), and +// - the CLI's GetAgentStatus snapshot (board.Snapshot). +// White-box (package runnerhub) so the test drives the unexported enroll/bind +// lifecycle directly. Sleep-free: the delivery records+fans synchronously under +// the projection's lock, so every assertion reads a settled fact. + +import ( + "context" + "testing" + "time" + + "github.com/RigelBuild/compass/go/events" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/board" + "github.com/RigelBuild/compass/go/internal/store" +) + +// newHubWithBoard wires the real Bridge board as the hub's LifecycleSink over a +// fresh bus, so a delivered lifecycle frame both fans onto SubscribeEvents and +// records into the snapshot. The bus is closed at test end. +func newHubOverBoard(t *testing.T) (*Hub, *board.Projection, *events.Bus[*compassv1.SubscribeEventsResponse]) { + t.Helper() + bus := events.NewBus[*compassv1.SubscribeEventsResponse]() + t.Cleanup(bus.Close) + brd := board.NewProjection(bus) + return NewHub(brd, &fakeTailSink{}, nil, discardLogger()), brd, bus +} + +// recvSessionStatus reads one live bus event and returns its AgentSessionStatus, +// failing fast on an early close or a stall. +func recvSessionStatus(t *testing.T, ch <-chan events.Stamped[*compassv1.SubscribeEventsResponse]) *compassv1.AgentSessionStatus { + t.Helper() + select { + case e, ok := <-ch: + if !ok { + t.Fatal("live channel closed before an event arrived") + } + got := e.Payload.GetAgentSessionStatus() + if got == nil { + t.Fatalf("live event carried a non-AgentSessionStatus payload: %v", e.Payload) + } + return got + case <-time.After(testTimeout): + t.Fatal("timed out waiting for a live event") + return nil + } +} + +// identitySessionID is the one session these tests drive; the tier/posture +// stamp is Runner-wide, so a second session would assert nothing new. +const identitySessionID = "sess-1" + +// deliverWorking pushes a WORKING lifecycle frame for the test session through +// the hub. +func deliverWorking(t *testing.T, hub *Hub, seq uint64) { + t.Helper() + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: seq, SessionID: identitySessionID, + Frame: sessionStateFrame(compassv1.AgentSessionState_AGENT_SESSION_STATE_WORKING), + }); err != nil { + t.Fatalf("Deliver(WORKING) = %v, want nil", err) + } +} + +// TestEnrolledTierAndPostureReachBothRenderPaths pins the central fix: a Runner +// enrolled as HOST/UNENFORCED has both facts stamped onto a session's status on +// BOTH surfaces — the SubscribeEvents fan-out (the UI path) and the board +// snapshot (the CLI's GetAgentStatus path) — off one delivery. +// +// Negative control: reverting deliverSession to publish {SessionId, State, +// AgentAccountId} with no tier/posture (or reverting runnerRuntimeIdentity to +// return UNSPECIFIED) reddens every assertion below — observed +// "bus event runtime_tier = RUNTIME_TIER_UNSPECIFIED, want RUNTIME_TIER_HOST". +func TestEnrolledTierAndPostureReachBothRenderPaths(t *testing.T) { + hub, brd, bus := newHubOverBoard(t) + + sub, err := bus.Subscribe(0, bus.InstanceEpoch()) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + t.Cleanup(sub.Cancel) + + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_HOST, compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED) + bindSession(hub, identitySessionID) // bind AFTER enroll (enroll clears bindings) + + deliverWorking(t, hub, 1) + + // UI path: the SubscribeEvents fan-out carries the enrolled tier + posture. + got := recvSessionStatus(t, sub.Live) + if got.GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_HOST { + t.Errorf("bus event runtime_tier = %v, want RUNTIME_TIER_HOST", got.GetRuntimeTier()) + } + if got.GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED { + t.Errorf("bus event egress_posture = %v, want EGRESS_POSTURE_UNENFORCED", got.GetEgressPosture()) + } + + // CLI path: the board snapshot (GetAgentStatus) carries them too. + snap := brd.Snapshot(identitySessionID) + if len(snap) != 1 { + t.Fatalf("Snapshot(sess-1) = %d entries, want 1", len(snap)) + } + if snap[0].GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_HOST { + t.Errorf("snapshot runtime_tier = %v, want RUNTIME_TIER_HOST", snap[0].GetRuntimeTier()) + } + if snap[0].GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED { + t.Errorf("snapshot egress_posture = %v, want EGRESS_POSTURE_UNENFORCED", snap[0].GetEgressPosture()) + } +} + +// TestReEnrollUpdatesStampedTierAndPosture pins the reattach invariant: a status +// published after a Runner re-enrolls with DIFFERENT values must reflect the +// NEWLY enrolled Runner's tier + posture, never the previous Runner's. A +// re-enroll clears bindings, so the session is re-bound before the second frame. +// +// Negative control: making runnerRuntimeIdentity read a cached first-enroll value +// (or dropping the h.mu read so it races the reattach) reddens the assertion — +// observed "runtime_tier = RUNTIME_TIER_HOST, want RUNTIME_TIER_PODMAN". +func TestReEnrollUpdatesStampedTierAndPosture(t *testing.T) { + hub, brd, _ := newHubOverBoard(t) + + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_HOST, compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED) + bindSession(hub, identitySessionID) + deliverWorking(t, hub, 1) + + // The Runner reconnects declaring a different tier + posture. + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_PODMAN, compassv1.EgressPosture_EGRESS_POSTURE_ARMED) + bindSession(hub, identitySessionID) + deliverWorking(t, hub, 2) + + snap := brd.Snapshot(identitySessionID) + if len(snap) != 1 { + t.Fatalf("Snapshot(sess-1) = %d entries, want 1", len(snap)) + } + if snap[0].GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_PODMAN { + t.Errorf("runtime_tier = %v, want RUNTIME_TIER_PODMAN (the re-enrolled Runner's value, not the stale HOST)", snap[0].GetRuntimeTier()) + } + if snap[0].GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_ARMED { + t.Errorf("egress_posture = %v, want EGRESS_POSTURE_ARMED (the re-enrolled Runner's value, not the stale UNENFORCED)", snap[0].GetEgressPosture()) + } +} + +// TestEnrollWithNoRuntimeIdentityYieldsUnspecified pins the fail-honest default: +// a Runner that declares nothing (zero values) yields UNSPECIFIED on both fields +// — never a plausible default like PODMAN or ARMED. A security surface that +// guesses is worse than one that says it does not know. +// +// Negative control: defaulting runnerRuntimeIdentity to PODMAN/ARMED when the +// enrolled values are zero reddens both assertions — observed "runtime_tier = +// RUNTIME_TIER_PODMAN, want RUNTIME_TIER_UNSPECIFIED". +func TestEnrollWithNoRuntimeIdentityYieldsUnspecified(t *testing.T) { + hub, brd, _ := newHubOverBoard(t) + + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) + bindSession(hub, identitySessionID) + deliverWorking(t, hub, 1) + + snap := brd.Snapshot(identitySessionID) + if len(snap) != 1 { + t.Fatalf("Snapshot(sess-1) = %d entries, want 1", len(snap)) + } + if snap[0].GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED { + t.Errorf("runtime_tier = %v, want RUNTIME_TIER_UNSPECIFIED (never a guessed default)", snap[0].GetRuntimeTier()) + } + if snap[0].GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED { + t.Errorf("egress_posture = %v, want EGRESS_POSTURE_UNSPECIFIED (never a guessed default)", snap[0].GetEgressPosture()) + } +} diff --git a/go/internal/runnerhub/secrets_test.go b/go/internal/runnerhub/secrets_test.go index 1df0fd830..35e96512b 100644 --- a/go/internal/runnerhub/secrets_test.go +++ b/go/internal/runnerhub/secrets_test.go @@ -20,6 +20,7 @@ import ( "connectrpc.com/connect" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/secrets" "github.com/RigelBuild/compass/go/internal/store" @@ -68,7 +69,7 @@ func runnerResolverForFetch() *fakeResolver { // can never pull the secret set. func TestFetchSecretsUnboundSessionPermissionDenied(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // No session bound: the hub has no live session for "sess-unbound". resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) @@ -92,7 +93,7 @@ func TestFetchSecretsUnboundSessionPermissionDenied(t *testing.T) { // delivery/kind enums translated at the edge. func TestFetchSecretsBoundSessionReturnsResolvedSet(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-1") resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{ {Name: "DB_URL", Value: "postgres://secret", Version: "v1", Delivery: secrets.DeliveryEnv, Kind: secrets.SecretGeneric}, @@ -126,7 +127,7 @@ func TestFetchSecretsBoundSessionReturnsResolvedSet(t *testing.T) { // swallowed as an empty set. func TestFetchSecretsResolveErrorInternal(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-1") resolver := &fakeResolverSecrets{resolveErr: errors.New("resolve boom")} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) @@ -146,7 +147,7 @@ func TestFetchSecretsResolveErrorInternal(t *testing.T) { // window, before any session) resolves the set via the container_name selector. func TestFetchSecretsByBoundContainerReturnsResolvedSet(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v", Version: "v1"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) @@ -166,7 +167,7 @@ func TestFetchSecretsByBoundContainerReturnsResolvedSet(t *testing.T) { // reached — the pre-exec analogue of the unbound-session rejection. func TestFetchSecretsUnboundContainerPermissionDenied(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) client := newRawRunnerClient(t, url, "runner-tok") @@ -184,7 +185,7 @@ func TestFetchSecretsUnboundContainerPermissionDenied(t *testing.T) { // is CodeInvalidArgument — a contract skew, never a silent empty set. func TestFetchSecretsMissingSelectorInvalidArgument(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) client := newRawRunnerClient(t, url, "runner-tok") @@ -203,7 +204,7 @@ func TestFetchSecretsMissingSelectorInvalidArgument(t *testing.T) { // ambiguous request (CodeInvalidArgument) rather than silently preferring one. func TestFetchSecretsBothSelectorsInvalidArgument(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) bindSession(hub, "sess-1") resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} @@ -227,7 +228,7 @@ func TestFetchSecretsBothSelectorsInvalidArgument(t *testing.T) { // without also tolerating a transient outage as "no secrets". func TestFetchSecretsNoResolverFailedPrecondition(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-1") url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, nil) client := newRawRunnerClient(t, url, "runner-tok") @@ -262,7 +263,7 @@ func TestResolvedSecretMappingRedactsValue(t *testing.T) { // content hash — an opaque counter. func TestSignalSecretsVersionPushesMonotonicToken(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-a") bindSession(hub, "sess-b") router, _, err := hub.routerFor("any") @@ -319,7 +320,7 @@ func TestSignalSecretsVersionPushesMonotonicToken(t *testing.T) { // notify is not an error. func TestSignalSecretsVersionNoLiveSessionsIsNoop(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") rec := newRecordingSend() router.attach(rec.send) diff --git a/go/internal/runtime/agent.go b/go/internal/runtime/agent.go index a6fdc1550..f9d3c8bb7 100644 --- a/go/internal/runtime/agent.go +++ b/go/internal/runtime/agent.go @@ -261,10 +261,7 @@ func (r *AgentRuntime) WriteAgentFile(ctx context.Context, id WorkloadID, uid ui // can surface it per session instead of inferring containment from a green // launch. func (r *AgentRuntime) EgressPosture() EgressPosture { - if r.egressUnenforced() { - return EgressPostureUnenforced - } - return EgressPostureArmed + return PostureOf(r.runtime) } // createAndStart creates then starts the container, cleaning up a created but diff --git a/go/internal/runtime/tier.go b/go/internal/runtime/tier.go index 37c367012..665a68413 100644 --- a/go/internal/runtime/tier.go +++ b/go/internal/runtime/tier.go @@ -36,6 +36,18 @@ func TierOf(engine WorkloadRuntime) WorkloadTier { return "" } +// PostureOf reports how a backend constrains agent egress, probing the same +// egressUnenforcer marker AgentRuntime.EgressPosture does so the value a Runner +// declares at enrollment matches the one a live workload would report. A backend +// with no isolation boundary to firewall is unenforced; every other backend is +// armed. +func PostureOf(engine WorkloadRuntime) EgressPosture { + if unenforcer, ok := engine.(egressUnenforcer); ok && unenforcer.EgressUnenforced() { + return EgressPostureUnenforced + } + return EgressPostureArmed +} + // Tier reports the podman tier. func (p *PodmanCLI) Tier() WorkloadTier { return WorkloadTierPodman } diff --git a/proto/compass/v1/runner.proto b/proto/compass/v1/runner.proto index 2d73f9bb3..68c5977a8 100644 --- a/proto/compass/v1/runner.proto +++ b/proto/compass/v1/runner.proto @@ -190,8 +190,15 @@ service RunnerService { // CodeUnauthenticated, so the field is a defense-in-depth cross-check, not a // trusted input. The credential itself rides the transport as a bearer token, // never a field here (mirroring IssueTokenRequest, compass.proto:237-242). +// +// runtime_tier and egress_posture are declared once here, not per session: +// they are Runner-wide facts of the one backend this Runner drives, so the hub +// stamps them onto every session it owns rather than have each lifecycle frame +// repeat them. message EnrollRequest { string runner_id = 1; + RuntimeTier runtime_tier = 2; + EgressPosture egress_posture = 3; } // Enroll response: the handshake ack. `reattached` distinguishes a fresh From 1811bb3fe299862a71e3540938d9a64594c4498f Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 12 Sep 2026 13:57:59 -0400 Subject: [PATCH 2/2] docs(runner): record the trust and nil-runner assumptions on enrollment Review findings on the enrollment-declared runtime identity, all documentation except the last: - EnrollRequest now says the tier and posture are trusted as declared. They sat directly beneath runner_id, whose doc spells out that it is cross-checked and untrusted, which invites the opposite inference or a cross-check that cannot exist: only the Runner can see which namespace an agent got, and it is the component enforcing egress. - deliverSession's comment no longer implies the account and the tier/posture are read atomically; they are two acquisitions and can straddle a re-enroll, which is harmless while one Runner enrolls at a time. - runnerRuntimeIdentity's nil branch records that no disconnect path nils h.runner, and that this is load-bearing: the board replaces its whole entry per publish, so a later UNSPECIFIED would regress a known tier. - The fail-honest test's docstring claimed more than the test gives. Its expectation is the field's zero value, so it pins only that a zero declaration stays zero; propagation is pinned elsewhere. - egressUnenforced now derives from PostureOf rather than repeating the marker probe. Co-authored-by: Matt Wilkinson --- go/internal/gen/compass/v1/runner.pb.go | 5 ++++- go/internal/runnerhub/hub.go | 12 +++++++----- go/internal/runnerhub/runtime_identity_test.go | 9 +++++---- go/internal/runtime/agent.go | 3 +-- proto/compass/v1/runner.proto | 5 ++++- 5 files changed, 21 insertions(+), 13 deletions(-) diff --git a/go/internal/gen/compass/v1/runner.pb.go b/go/internal/gen/compass/v1/runner.pb.go index c2a630358..4b4c68969 100644 --- a/go/internal/gen/compass/v1/runner.pb.go +++ b/go/internal/gen/compass/v1/runner.pb.go @@ -133,7 +133,10 @@ func (RunnerErrorCode) EnumDescriptor() ([]byte, []int) { // runtime_tier and egress_posture are declared once here, not per session: // they are Runner-wide facts of the one backend this Runner drives, so the hub // stamps them onto every session it owns rather than have each lifecycle frame -// repeat them. +// repeat them. Unlike runner_id they are trusted as declared, with nothing to +// cross-check them against: only the Runner can observe which namespace an +// agent got, and it is the component that enforces egress in the first place, +// so a Runner that lies here has already lost containment either way. type EnrollRequest struct { state protoimpl.MessageState `protogen:"open.v1"` RunnerId string `protobuf:"bytes,1,opt,name=runner_id,json=runnerId,proto3" json:"runner_id,omitempty"` diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index f6d978190..362b7be00 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -805,8 +805,9 @@ func (h *Hub) deliverSession(ctx context.Context, sessionID string, sf *compassv // gap). accountForSession takes h.mu; deliverSession holds no lock here. // runnerRuntimeIdentity reads the enrolled Runner's tier/posture under the // same lock, so a status published after a reattach reflects the newly - // enrolled Runner's values. It is a separate critical section from the - // account resolve above, not one atomic read of both. + // enrolled Runner's values, never the previous Runner's. It is a separate + // acquisition from the account read above, so the two can straddle a + // re-enroll; harmless while one Runner enrolls at a time. account, hasAccount := h.accountForSession(ctx, sessionID) tier, egressPosture := h.runnerRuntimeIdentity() status := &compassv1.AgentSessionStatus{SessionId: sessionID, State: state, RuntimeTier: tier, EgressPosture: egressPosture} @@ -1166,13 +1167,14 @@ func (h *Hub) routerFor(sessionID string) (*commandRouter, string, error) { // runnerRuntimeIdentity returns the enrolled Runner's declared runtime tier and // egress posture under h.mu, so a session status stamps the Runner that owns it // today rather than racing a re-enroll. No Runner enrolled yields UNSPECIFIED on -// both — the wire's "we do not know", never a plausible default. h.runner is -// assigned only in enroll and never set back to nil, so no disconnect path can -// regress a known tier to UNSPECIFIED. +// both — the wire's "we do not know", never a plausible default. func (h *Hub) runnerRuntimeIdentity() (compassv1.RuntimeTier, compassv1.EgressPosture) { h.mu.Lock() defer h.mu.Unlock() if h.runner == nil { + // Reachable only before the first enroll: no disconnect path nils + // h.runner, and that is load-bearing — the board replaces its whole + // entry per publish, so a later UNSPECIFIED would regress a known tier. return compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED } return h.runner.tier, h.runner.egressPosture diff --git a/go/internal/runnerhub/runtime_identity_test.go b/go/internal/runnerhub/runtime_identity_test.go index 72d770a73..9d0d70005 100644 --- a/go/internal/runnerhub/runtime_identity_test.go +++ b/go/internal/runnerhub/runtime_identity_test.go @@ -152,10 +152,11 @@ func TestReEnrollUpdatesStampedTierAndPosture(t *testing.T) { } } -// TestEnrollWithNoRuntimeIdentityYieldsUnspecified pins the fail-honest default: -// a Runner that declares nothing (zero values) yields UNSPECIFIED on both fields -// — never a plausible default like PODMAN or ARMED. A security surface that -// guesses is worse than one that says it does not know. +// TestEnrollWithNoRuntimeIdentityYieldsUnspecified pins runnerRuntimeIdentity's +// fail-honest default only: a Runner that declares nothing stays UNSPECIFIED +// rather than a plausible PODMAN/ARMED, because a security surface that guesses +// is worse than one that says it does not know. Its expectation is the zero +// value, so propagation is pinned by the two tests above, not here. // // Negative control: defaulting runnerRuntimeIdentity to PODMAN/ARMED when the // enrolled values are zero reddens both assertions — observed "runtime_tier = diff --git a/go/internal/runtime/agent.go b/go/internal/runtime/agent.go index f9d3c8bb7..5536adf92 100644 --- a/go/internal/runtime/agent.go +++ b/go/internal/runtime/agent.go @@ -373,8 +373,7 @@ func (r *AgentRuntime) provision(ctx context.Context, id WorkloadID, spec AgentS } func (r *AgentRuntime) egressUnenforced() bool { - unenforcer, ok := r.runtime.(egressUnenforcer) - return ok && unenforcer.EgressUnenforced() + return PostureOf(r.runtime) == EgressPostureUnenforced } func (r *AgentRuntime) selfArmsEgress() bool { diff --git a/proto/compass/v1/runner.proto b/proto/compass/v1/runner.proto index 68c5977a8..ae23664c1 100644 --- a/proto/compass/v1/runner.proto +++ b/proto/compass/v1/runner.proto @@ -194,7 +194,10 @@ service RunnerService { // runtime_tier and egress_posture are declared once here, not per session: // they are Runner-wide facts of the one backend this Runner drives, so the hub // stamps them onto every session it owns rather than have each lifecycle frame -// repeat them. +// repeat them. Unlike runner_id they are trusted as declared, with nothing to +// cross-check them against: only the Runner can observe which namespace an +// agent got, and it is the component that enforces egress in the first place, +// so a Runner that lies here has already lost containment either way. message EnrollRequest { string runner_id = 1; RuntimeTier runtime_tier = 2;