automations: feat: add shared automations protocol

clients/go/ahp/client_test.go

@@ -607,6 +607,79 @@ func TestClientSubscriptionFanOut(t *testing.T) {
 	}
 }
 
+func TestClientAutomationCatalogueAction(t *testing.T) {
+	clientSide, serverSide := newMemTransportPair()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	client, err := Connect(ctx, clientSide, DefaultConfig())
+	if err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	defer client.Shutdown(context.Background())
+
+	const automationsURI = "ahp-automations://"
+	sub := client.AttachSubscription(automationsURI)
+	stream := client.Events()
+	automationURI := ahptypes.URI("ahp-automation:/nightly")
+
+	params, err := json.Marshal(ahptypes.ActionEnvelope{
+		Channel:   automationsURI,
+		ServerSeq: 1,
+		Action: ahptypes.StateAction{
+			Value: &ahptypes.AutomationRemovedAction{
+				Type:     ahptypes.ActionTypeAutomationRemoved,
+				Resource: automationURI,
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("marshal action: %v", err)
+	}
+	wire, err := EncodeMessage(ahptypes.JsonRpcMessage{Notification: &ahptypes.JsonRpcNotification{
+		JsonRpc: ahptypes.JsonRpcV2,
+		Method:  "action",
+		Params:  params,
+	}})
+	if err != nil {
+		t.Fatalf("encode notification: %v", err)
+	}
+	if err := serverSide.Send(ctx, wire); err != nil {
+		t.Fatalf("send notification: %v", err)
+	}
+
+	check := func(t *testing.T, event SubscriptionEvent) {
+		t.Helper()
+		action, ok := event.(SubscriptionEventAction)
+		if !ok {
+			t.Fatalf("got %T, want SubscriptionEventAction", event)
+		}
+		removed, ok := action.Envelope.Action.Value.(*ahptypes.AutomationRemovedAction)
+		if !ok {
+			t.Fatalf("got %T, want AutomationRemovedAction", action.Envelope.Action.Value)
+		}
+		if removed.Resource != automationURI {
+			t.Errorf("resource = %q, want %q", removed.Resource, automationURI)
+		}
+	}
+
+	select {
+	case event := <-sub.Events():
+		check(t, event)
+	case <-ctx.Done():
+		t.Fatal("subscription did not receive event")
+	}
+
+	select {
+	case event := <-stream.Events():
+		if event.Channel != automationsURI {
+			t.Errorf("channel = %q, want %q", event.Channel, automationsURI)
+		}
+		check(t, event.Event)
+	case <-ctx.Done():
+		t.Fatal("top-level stream did not receive event")
+	}
+}
+
 // TestClientShutdownFailsInFlightRequest confirms a Shutdown unblocks
 // any pending request with ErrShutdown.
 func TestClientShutdownFailsInFlightRequest(t *testing.T) {

clients/go/ahp/hosts/hosts.go

@@ -213,6 +213,7 @@ type HostHandle struct {
 	ClientID        string
 	State           HostState
 	ProtocolVersion string
+	Automations     *ahptypes.AutomationCapabilities
 	Agents          []ahptypes.AgentInfo
 	Sessions        []ahptypes.SessionSummary
 	Terminals       []ahptypes.TerminalInfo
@@ -400,21 +401,22 @@ var ErrDuplicateHost = errors.New("hosts: host id already registered")
 
 // hostState is the per-host bookkeeping the multi-host runtime owns.
 type hostState struct {
-	id         HostID
-	label      string
-	cfg        HostConfig
-	mu         sync.RWMutex
-	client     *ahp.Client
-	state      HostState
-	clientID   string
-	protoVer   string
-	agents     []ahptypes.AgentInfo
-	sessions   []ahptypes.SessionSummary
-	terminals  []ahptypes.TerminalInfo
-	updatedAt  time.Time
-	generation uint64
-	cancel     context.CancelFunc
-	supervised sync.WaitGroup
+	id          HostID
+	label       string
+	cfg         HostConfig
+	mu          sync.RWMutex
+	client      *ahp.Client
+	state       HostState
+	clientID    string
+	protoVer    string
+	automations *ahptypes.AutomationCapabilities
+	agents      []ahptypes.AgentInfo
+	sessions    []ahptypes.SessionSummary
+	terminals   []ahptypes.TerminalInfo
+	updatedAt   time.Time
+	generation  uint64
+	cancel      context.CancelFunc
+	supervised  sync.WaitGroup
 }
 
 // MultiHostClient is the public multi-host registry + reconnect
@@ -573,6 +575,7 @@ func (m *MultiHostClient) openHost(ctx context.Context, hs *hostState) error {
 	hs.mu.Lock()
 	hs.client = client
 	hs.protoVer = result.ProtocolVersion
+	hs.automations = cloneAutomationCapabilities(result.Automations)
 	hs.generation++
 	hs.mu.Unlock()
 
@@ -716,13 +719,43 @@ func (m *MultiHostClient) snapshotHandle(hs *hostState) *HostHandle {
 		ClientID:        hs.clientID,
 		State:           hs.state,
 		ProtocolVersion: hs.protoVer,
+		Automations:     cloneAutomationCapabilities(hs.automations),
 		Agents:          append([]ahptypes.AgentInfo(nil), hs.agents...),
 		Sessions:        append([]ahptypes.SessionSummary(nil), hs.sessions...),
 		Terminals:       append([]ahptypes.TerminalInfo(nil), hs.terminals...),
 		UpdatedAt:       hs.updatedAt,
 	}
 }
 
+func cloneAutomationCapabilities(capabilities *ahptypes.AutomationCapabilities) *ahptypes.AutomationCapabilities {
+	if capabilities == nil {
+		return nil
+	}
+
+	clone := *capabilities
+	if capabilities.Create != nil {
+		value := *capabilities.Create
+		clone.Create = &value
+	}
+	if capabilities.Schedules != nil {
+		value := *capabilities.Schedules
+		if capabilities.Schedules.MinIntervalMinutes != nil {
+			minIntervalMinutes := *capabilities.Schedules.MinIntervalMinutes
+			value.MinIntervalMinutes = &minIntervalMinutes
+		}
+		clone.Schedules = &value
+	}
+	if capabilities.RunCancellation != nil {
+		value := *capabilities.RunCancellation
+		clone.RunCancellation = &value
+	}
+	if capabilities.RunHistoryLimit != nil {
+		value := *capabilities.RunHistoryLimit
+		clone.RunHistoryLimit = &value
+	}
+	return &clone
+}
+
 // ClientHandle returns a generation-checked [HostClientHandle] for
 // the named host, or ErrUnknownHost if the host is not registered.
 func (m *MultiHostClient) ClientHandle(id HostID) (*HostClientHandle, error) {

clients/go/ahp/hosts/hosts_test.go

@@ -66,6 +66,12 @@ func (t *fakeTransport) Close(_ context.Context) error {
 // runFakeServer responds to one Initialize request with a stub
 // InitializeResult. It exits when the transport closes.
 func runFakeServer(t *testing.T, serverSide *fakeTransport) {
+	runFakeServerWithInitializeResult(t, serverSide, ahptypes.InitializeResult{
+		ProtocolVersion: ahptypes.ProtocolVersion,
+	})
+}
+
+func runFakeServerWithInitializeResult(t *testing.T, serverSide *fakeTransport, initializeResult ahptypes.InitializeResult) {
 	t.Helper()
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 	defer cancel()
@@ -82,7 +88,7 @@ func runFakeServer(t *testing.T, serverSide *fakeTransport) {
 			continue
 		}
 		if parsed.Request.Method == "initialize" {
-			result, _ := json.Marshal(ahptypes.InitializeResult{ProtocolVersion: ahptypes.ProtocolVersion})
+			result, _ := json.Marshal(initializeResult)
 			resp := ahptypes.JsonRpcMessage{SuccessResponse: &ahptypes.JsonRpcSuccessResponse{
 				JsonRpc: ahptypes.JsonRpcV2,
 				ID:      parsed.Request.ID,
@@ -94,6 +100,72 @@ func runFakeServer(t *testing.T, serverSide *fakeTransport) {
 	}
 }
 
+func TestAutomationCapabilitiesUpdatedAcrossReconnect(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	multi := NewMultiHostClient()
+	defer multi.Shutdown(context.Background())
+
+	servers := make(chan *fakeTransport, 2)
+	attempt := 0
+	cfg := NewHostConfig("automation-host", "Automation Host", func(_ context.Context, _ HostID) (ahp.Transport, error) {
+		attempt++
+		clientSide, serverSide := newFakePair()
+		runHistoryLimit := int64(10)
+		if attempt > 1 {
+			runHistoryLimit = 25
+		}
+		go runFakeServerWithInitializeResult(t, serverSide, ahptypes.InitializeResult{
+			ProtocolVersion: ahptypes.ProtocolVersion,
+			Automations: &ahptypes.AutomationCapabilities{
+				RunHistoryLimit: &runHistoryLimit,
+			},
+		})
+		servers <- serverSide
+		return clientSide, nil
+	})
+	cfg.ReconnectPolicy = ReconnectPolicy{
+		MaxAttempts:       2,
+		InitialBackoff:    time.Millisecond,
+		MaxBackoff:        time.Millisecond,
+		BackoffMultiplier: 1,
+		ResetOnSuccess:    true,
+	}
+
+	handle, err := multi.AddHost(ctx, cfg)
+	if err != nil {
+		t.Fatalf("AddHost: %v", err)
+	}
+	if handle.Automations == nil {
+		t.Fatal("initial Automations is nil")
+	}
+	if got := handle.Automations.RunHistoryLimit; got == nil || *got != 10 {
+		t.Fatalf("initial run history limit = %v, want 10", got)
+	}
+
+	firstServer := <-servers
+	if err := firstServer.Close(ctx); err != nil {
+		t.Fatalf("close first server: %v", err)
+	}
+
+	for {
+		handle = multi.Host(cfg.ID)
+		if handle != nil &&
+			handle.State.Kind == HostStateConnected &&
+			handle.Automations != nil &&
+			handle.Automations.RunHistoryLimit != nil &&
+			*handle.Automations.RunHistoryLimit == 25 {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			t.Fatal("automation capabilities were not updated after reconnect")
+		case <-time.After(time.Millisecond):
+		}
+	}
+}
+
 // TestSingleHostHandshake exercises the [Single] one-line constructor
 // against a fake server and confirms the host transitions to the
 // Connected state with a populated protocol version.