Add remote poll message pump

Author: SqlRush

cmd/claude/main.go

@@ -30,6 +30,7 @@ import (
 	"ccgo/internal/model"
 	"ccgo/internal/permissions"
 	pluginpkg "ccgo/internal/plugins"
+	remotepkg "ccgo/internal/remote"
 	"ccgo/internal/session"
 	"ccgo/internal/tool"
 	filetools "ccgo/internal/tools/file"
@@ -698,7 +699,7 @@ func runDaemon(ctx context.Context, state *bootstrap.State, options daemonOption
 		if err := writeHeartbeat(now); err != nil {
 			return contracts.ToolResult{}, err
 		}
-		return runDaemonDueSchedules(ctx, runner, now)
+		return runDaemonTick(ctx, runner, now)
 	}
 	if !options.Once {
 		daemonServer, err = daemonpkg.StartServer(daemonpkg.ServerOptions{
@@ -774,6 +775,172 @@ func runDaemonDueSchedules(ctx context.Context, runner conversation.Runner, now
 	}, "", now, tool.NopProgressSink())
 }
 
+func runDaemonTick(ctx context.Context, runner conversation.Runner, now time.Time) (contracts.ToolResult, error) {
+	scheduleResult, err := runDaemonDueSchedules(ctx, runner, now)
+	if err != nil {
+		return contracts.ToolResult{}, err
+	}
+	remoteResult := runDaemonRemotePoll(ctx, runner, now)
+	schedule := scheduleResult.StructuredContent
+	remotePoll := remoteResult.StructuredContent
+	scheduleTriggered := intMapValue(schedule, "triggered_count")
+	scheduleErrors := intMapValue(schedule, "error_count")
+	remoteDelivered := intMapValue(remotePoll, "delivered_count")
+	remoteErrors := intMapValue(remotePoll, "error_count")
+	structured := map[string]any{
+		"type":                        "daemon_tick",
+		"checked_at":                  now.UTC().Format(time.RFC3339Nano),
+		"due_count":                   intMapValue(schedule, "due_count"),
+		"triggered_count":             scheduleTriggered + remoteDelivered,
+		"error_count":                 scheduleErrors + remoteErrors,
+		"schedule_triggered_count":    scheduleTriggered,
+		"schedule_error_count":        scheduleErrors,
+		"remote_delivered_count":      remoteDelivered,
+		"remote_poll_error_count":     remoteErrors,
+		"remote_poll_duplicate_count": intMapValue(remotePoll, "duplicate_count"),
+		"schedule":                    schedule,
+		"remote_poll":                 remotePoll,
+	}
+	return contracts.ToolResult{
+		Content:           fmt.Sprintf("Triggered %d due schedule(s), delivered %d remote event(s); %d error(s).", scheduleTriggered, remoteDelivered, scheduleErrors+remoteErrors),
+		StructuredContent: structured,
+	}, nil
+}
+
+func runDaemonRemotePoll(ctx context.Context, runner conversation.Runner, now time.Time) contracts.ToolResult {
+	structured := map[string]any{
+		"type":            "remote_poll",
+		"checked_at":      now.UTC().Format(time.RFC3339Nano),
+		"runtime_state":   remotepkg.PumpDisabled,
+		"event_count":     0,
+		"delivered_count": 0,
+		"duplicate_count": 0,
+		"error_count":     0,
+	}
+	registrationPath := remotepkg.SessionRegistrationPath(runner.SessionPath, runner.SessionID)
+	pumpPath := remotepkg.SessionPumpPath(runner.SessionPath, runner.SessionID)
+	if registrationPath == "" || pumpPath == "" {
+		structured["error_count"] = 1
+		structured["error"] = "remote registration path is unavailable"
+		return contracts.ToolResult{Content: "Remote poll is not configured.", StructuredContent: structured}
+	}
+	registration, err := remotepkg.LoadRegistrationState(registrationPath)
+	if err != nil {
+		structured["runtime_state"] = remotepkg.PumpFailed
+		structured["error_count"] = 1
+		structured["error"] = err.Error()
+		_ = remotepkg.WritePumpState(pumpPath, remotepkg.PumpState{
+			SessionID:    runner.SessionID,
+			RuntimeState: remotepkg.PumpFailed,
+			LastPollAt:   now.UTC().Format(time.RFC3339Nano),
+			ErrorCount:   1,
+			LastError:    err.Error(),
+		})
+		return contracts.ToolResult{Content: "Remote poll failed.", StructuredContent: structured}
+	}
+	if registration.RuntimeState != remotepkg.RegistrationRegistered || strings.TrimSpace(registration.PollURL) == "" {
+		_ = remotepkg.WritePumpState(pumpPath, remotepkg.PumpState{
+			SessionID:    runner.SessionID,
+			RuntimeState: remotepkg.PumpDisabled,
+			PollURL:      remotepkg.DisplayEndpoint(registration.PollURL),
+			LastPollAt:   now.UTC().Format(time.RFC3339Nano),
+		})
+		return contracts.ToolResult{Content: "Remote poll is disabled.", StructuredContent: structured}
+	}
+	previous, err := remotepkg.LoadPumpState(pumpPath)
+	if err != nil {
+		previous.LastError = err.Error()
+	}
+	settings := runner.MergedSettings()
+	authToken := ""
+	if settings.Remote != nil {
+		authToken = settings.Remote.AuthToken
+	}
+	poll := remotepkg.FetchPollEvents(ctx, remotepkg.PollOptions{
+		PollURL:   registration.PollURL,
+		Cursor:    previous.LastCursor,
+		AuthToken: authToken,
+	})
+	pumpState := remotepkg.PumpState{
+		SessionID:    runner.SessionID,
+		RuntimeState: remotepkg.PumpRunning,
+		PollURL:      remotepkg.DisplayEndpoint(registration.PollURL),
+		LastCursor:   previous.LastCursor,
+		LastPollAt:   now.UTC().Format(time.RFC3339Nano),
+		StatusCode:   poll.StatusCode,
+		EventCount:   len(poll.Events),
+	}
+	if poll.NextCursor != "" {
+		pumpState.LastCursor = poll.NextCursor
+	}
+	if poll.Error != "" {
+		pumpState.RuntimeState = remotepkg.PumpFailed
+		pumpState.ErrorCount = 1
+		pumpState.LastError = poll.Error
+		structured["runtime_state"] = pumpState.RuntimeState
+		structured["status_code"] = poll.StatusCode
+		structured["event_count"] = len(poll.Events)
+		structured["error_count"] = 1
+		structured["error"] = poll.Error
+		_ = remotepkg.WritePumpState(pumpPath, pumpState)
+		return contracts.ToolResult{Content: "Remote poll failed.", StructuredContent: structured}
+	}
+	errorsOut := make([]map[string]any, 0)
+	delivered := 0
+	duplicates := 0
+	for _, event := range poll.Events {
+		input, err := json.Marshal(map[string]any{
+			"team_id":  event.TeamID,
+			"target":   event.Target,
+			"event_id": event.EventID,
+			"source":   event.Source,
+			"event":    event.Event,
+			"message":  event.Message,
+		})
+		if err != nil {
+			errorsOut = append(errorsOut, map[string]any{"event_id": event.EventID, "error": err.Error()})
+			continue
+		}
+		result, err := tasktools.RunRemoteTrigger(tool.Context{
+			Context:          ctx,
+			WorkingDirectory: runner.WorkingDirectory,
+			SessionID:        runner.SessionID,
+			Metadata: map[string]any{
+				tool.MetadataSessionPathKey: runner.SessionPath,
+			},
+		}, input, tool.NopProgressSink())
+		if err != nil {
+			errorsOut = append(errorsOut, map[string]any{"event_id": event.EventID, "team_id": event.TeamID, "error": err.Error()})
+			continue
+		}
+		if duplicate, _ := result.StructuredContent["duplicate"].(bool); duplicate {
+			duplicates++
+			continue
+		}
+		delivered += intMapValue(result.StructuredContent, "sent_count")
+	}
+	pumpState.DeliveredCount = delivered
+	pumpState.DuplicateCount = duplicates
+	pumpState.ErrorCount = len(errorsOut)
+	if len(errorsOut) > 0 {
+		pumpState.LastError = fmt.Sprint(errorsOut[0]["error"])
+	}
+	_ = remotepkg.WritePumpState(pumpPath, pumpState)
+	structured["runtime_state"] = pumpState.RuntimeState
+	structured["poll_url"] = pumpState.PollURL
+	structured["last_cursor"] = pumpState.LastCursor
+	structured["status_code"] = pumpState.StatusCode
+	structured["event_count"] = pumpState.EventCount
+	structured["delivered_count"] = pumpState.DeliveredCount
+	structured["duplicate_count"] = pumpState.DuplicateCount
+	structured["error_count"] = pumpState.ErrorCount
+	structured["errors"] = errorsOut
+	return contracts.ToolResult{
+		Content:           fmt.Sprintf("Remote poll delivered %d event(s); %d duplicate(s); %d error(s).", delivered, duplicates, len(errorsOut)),
+		StructuredContent: structured,
+	}
+}
+
 func daemonTickResponse(result contracts.ToolResult) daemonpkg.TickResponse {
 	structured := result.StructuredContent
 	return daemonpkg.TickResponse{

cmd/claude/main_test.go

@@ -21,6 +21,7 @@ import (
 	daemonpkg "ccgo/internal/daemon"
 	integrationspkg "ccgo/internal/integrations"
 	"ccgo/internal/messages"
+	remotepkg "ccgo/internal/remote"
 	"ccgo/internal/session"
 	"ccgo/internal/tool"
 	bashtools "ccgo/internal/tools/bash"
@@ -301,6 +302,87 @@ func TestRunDaemonDueSchedulesNoopsWithoutSchedules(t *testing.T) {
 	}
 }
 
+func TestRunDaemonRemotePollInjectsRemoteTriggers(t *testing.T) {
+	dir := t.TempDir()
+	transcriptPath := filepath.Join(dir, "session.jsonl")
+	sessionID := contracts.ID("sess_daemon_remote")
+	manager := session.NewSidechainManager(transcriptPath, sessionID)
+	if _, err := manager.Start(session.SidechainOptions{ID: "agent/remote-lead", StartedAt: time.Unix(100, 0).UTC()}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := manager.Start(session.SidechainOptions{ID: "agent/remote-member", StartedAt: time.Unix(101, 0).UTC()}); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := manager.CreateTeam(session.TeamOptions{
+		ID:                "remote/team",
+		CoordinatorTaskID: "agent/remote-lead",
+		TaskIDs:           []string{"agent/remote-member"},
+		Timestamp:         time.Unix(102, 0).UTC(),
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var cursors []string
+	var auths []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		cursors = append(cursors, r.URL.Query().Get("cursor"))
+		auths = append(auths, r.Header.Get("Authorization"))
+		w.Header().Set("content-type", "application/json")
+		_, _ = w.Write([]byte(`{"next_cursor":"cursor-2","events":[{"id":"delivery-1","team":"remote/team","source":"webhook","event":"deploy","message":"Deploy now."}]}`))
+	}))
+	defer server.Close()
+	if err := remotepkg.WriteRegistrationState(remotepkg.SessionRegistrationPath(transcriptPath, sessionID), remotepkg.RegistrationState{
+		SessionID:       sessionID,
+		RuntimeState:    remotepkg.RegistrationRegistered,
+		PollURL:         server.URL + "/poll?token=secret",
+		RemoteSessionID: "remote-session",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	runner := conversation.Runner{
+		SessionID:        sessionID,
+		SessionPath:      transcriptPath,
+		WorkingDirectory: dir,
+		MCP: &conversation.MCPConfig{UserSettings: contracts.Settings{
+			Remote: &contracts.RemoteSetting{AuthToken: "poll-token"},
+		}},
+	}
+	first := runDaemonRemotePoll(context.Background(), runner, time.Unix(200, 0).UTC())
+	if first.StructuredContent["runtime_state"] != remotepkg.PumpRunning || first.StructuredContent["delivered_count"] != 1 || first.StructuredContent["duplicate_count"] != 0 || first.StructuredContent["error_count"] != 0 {
+		t.Fatalf("first poll = %#v", first.StructuredContent)
+	}
+	if len(cursors) != 1 || cursors[0] != "" || auths[0] != "Bearer poll-token" {
+		t.Fatalf("first cursor/auth = %#v %#v", cursors, auths)
+	}
+	pump, err := remotepkg.LoadPumpState(remotepkg.SessionPumpPath(transcriptPath, sessionID))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if pump.LastCursor != "cursor-2" || pump.PollURL != server.URL+"/poll" || pump.DeliveredCount != 1 {
+		t.Fatalf("pump = %#v", pump)
+	}
+	resume, err := manager.ResumeContext("agent/remote-lead", 3)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(resume.Messages) != 2 || !strings.Contains(messages.TextContent(resume.Messages[1]), "Remote trigger received.") || !strings.Contains(messages.TextContent(resume.Messages[1]), "Deploy now.") {
+		t.Fatalf("resume messages = %#v", resume.Messages)
+	}
+	second := runDaemonRemotePoll(context.Background(), runner, time.Unix(201, 0).UTC())
+	if second.StructuredContent["delivered_count"] != 0 || second.StructuredContent["duplicate_count"] != 1 || second.StructuredContent["error_count"] != 0 {
+		t.Fatalf("second poll = %#v", second.StructuredContent)
+	}
+	if len(cursors) != 2 || cursors[1] != "cursor-2" {
+		t.Fatalf("cursors = %#v", cursors)
+	}
+	resume, err = manager.ResumeContext("agent/remote-lead", 5)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(resume.Messages) != 2 {
+		t.Fatalf("duplicate should not append, messages = %#v", resume.Messages)
+	}
+}
+
 func TestRunDaemonServesHealthEndpoint(t *testing.T) {
 	t.Setenv("CLAUDE_CONFIG_DIR", t.TempDir())
 	cwd := t.TempDir()