diff --git a/README.md b/README.md index 7327e43..61d4534 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,56 @@ and then you can start as many instances you want, each of them are isolated: vix ``` +### Remote control + +`vixd` can optionally accept prompts from Telegram or WhatsApp and reply with the +agent's answer. Remote control is disabled by default and requires both a feature +flag and an allowlist in `~/.vix/settings.json`: + +```json +{ + "features": { "remote_control": true }, + "remote_control": { + "enabled": true, + "cwd": "/absolute/project/path", + "workflow": "", + "max_concurrent_runs": 1, + "telegram": { + "enabled": true, + "bot_token": "", + "allowed_chat_ids": ["123456789"] + }, + "whatsapp": { + "enabled": true, + "access_token": "", + "app_secret": "", + "phone_number_id": "", + "verify_token": "", + "graph_api_version": "v20.0", + "webhook_addr": "127.0.0.1:1340", + "allowed_contacts": ["15551234567"] + } + } +} +``` + +Telegram uses bot long-polling. WhatsApp exposes `GET/POST /whatsapp` on +`webhook_addr` for Cloud API webhook verification and messages. Only allowlisted +chat IDs/contacts can control vix. Leave `workflow` empty for a plain prompt, or +set it to a workflow name to run that workflow for each remote prompt. + +Remote control is not a sandbox guarantee. A remote prompt uses the same +agent/tooling path as vix, subject to unattended confirmation denial and the +configured policy: the configured `cwd`, `$HOME`, and platform system paths may +be auto-allowed, and output is sent back through the chat provider. Treat a +compromised allowlisted account, or provider credentials capable of injecting +trusted inbound messages, as host access within that policy. Treat bot/provider +tokens as sensitive because they can expose traffic, disrupt service, or combine +with sender compromise. Use a locked-down `cwd`, keep `max_concurrent_runs` low, +and add deny-list entries for sensitive paths such as `~/.ssh`, `~/.aws`, +`~/.kube`, password-manager data, and cloud credential directories. Remote runs +deny confirmation prompts and cannot answer user questions or approve plans. +
## Providers & Configuration diff --git a/cmd/vixd/main.go b/cmd/vixd/main.go index e62659f..eb5fd31 100644 --- a/cmd/vixd/main.go +++ b/cmd/vixd/main.go @@ -194,6 +194,18 @@ func main() { server.RegisterHandler(cmd, handler) }, cred, ctx) daemon.RegisterToolHandlers(server) + if config.RemoteControlEnabled() { + remoteCfg, err := daemon.LoadRemoteControlConfig() + if err != nil { + log.Printf("remote control: config load failed: %v", err) + } else if remoteCfg.Enabled { + if err := server.StartRemoteControl(ctx, remoteCfg); err != nil { + log.Printf("remote control: disabled: %v", err) + } + } + } else { + log.Printf("remote control: disabled (features.remote_control=false or VIX_DISABLE_REMOTE_CONTROL)") + } if *webPort > 0 && !*noMissionControl { go daemon.StartWebServer(ctx, server, *webPort) } diff --git a/e2e/scenarios/remote_control_test.go b/e2e/scenarios/remote_control_test.go new file mode 100644 index 0000000..6711b6c --- /dev/null +++ b/e2e/scenarios/remote_control_test.go @@ -0,0 +1,26 @@ +package scenarios + +import ( + "testing" + + "github.com/get-vix/vix/e2e/harness" +) + +// TestRemoteControlUnattendedPolicyAcceptance is a staged acceptance spec for +// remote-control daemon sessions. The current harness can drive TUI sessions and +// CLI-triggered jobs/hooks, but it has no provider ingress/reply primitive for +// Telegram or WhatsApp. +// +// When enabled it should inject a trusted remote message and assert that the +// resulting vix-initiated session denies tool confirmation requests, returns an +// error for user questions and plan proposals, and replies to the provider with +// the final text or remote-control error. +func TestRemoteControlUnattendedPolicyAcceptance(t *testing.T) { + meta := harness.Meta{ + Category: "remote_control", + Subcategory: "remote_control.unattended_policy", + Description: "remote-control runs deny confirmations and fail closed for interactive question/plan events", + Wire: harness.WireMessages, + } + harness.SkipScenario(t, meta, "remote-control provider ingress/reply injection is not exposed by the current e2e harness; daemon policy is covered by internal/daemon unit tests") +} diff --git a/internal/config/config.go b/internal/config/config.go index 88d1fec..832b3fa 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -237,6 +237,16 @@ func HooksEnabled() bool { return feature("hooks", true) } +// RemoteControlEnabled reads the remote_control feature flag. Defaults to false +// because chat-service control requires explicit credentials and sender allowlists. +// VIX_DISABLE_REMOTE_CONTROL is an emergency kill switch. +func RemoteControlEnabled() bool { + if v := os.Getenv("VIX_DISABLE_REMOTE_CONTROL"); v == "1" || v == "true" { + return false + } + return feature("remote_control", false) +} + // OAuthPlaintextFallback reads the oauth_plaintext_fallback feature flag. It is // an opt-in escape hatch (default false): when true, interactive OAuth logins // may persist their tokens to the plaintext auth.json (0600) on machines with no diff --git a/internal/config/defaults/settings.json b/internal/config/defaults/settings.json index 6cbbbdd..33acaf9 100644 --- a/internal/config/defaults/settings.json +++ b/internal/config/defaults/settings.json @@ -5,7 +5,29 @@ "read_claude_md": true, "show_thinking": false, "telemetry": true, - "tool_orchestrator": false + "tool_orchestrator": false, + "remote_control": false + }, + "remote_control": { + "enabled": false, + "cwd": "", + "workflow": "", + "max_concurrent_runs": 1, + "telegram": { + "enabled": false, + "bot_token": "", + "allowed_chat_ids": [] + }, + "whatsapp": { + "enabled": false, + "access_token": "", + "app_secret": "", + "phone_number_id": "", + "verify_token": "", + "graph_api_version": "v20.0", + "webhook_addr": "127.0.0.1:1340", + "allowed_contacts": [] + } }, "elevenlabs": { "agent_id": "agent_7501kqrztj1te17ssqz5wqpnvkf3", diff --git a/internal/daemon/hook_runner.go b/internal/daemon/hook_runner.go index 6b4bddc..0711b3f 100644 --- a/internal/daemon/hook_runner.go +++ b/internal/daemon/hook_runner.go @@ -267,12 +267,12 @@ consume: case ev := <-session.eventChan: switch ev.Type { case "event.stream_chunk": - finalText.WriteString(decodeJobEvent[protocol.EventStreamChunk](ev.Data).Text) + finalText.WriteString(decodeUnattendedEvent[protocol.EventStreamChunk](ev.Data).Text) case "event.confirm_request": data, _ := json.Marshal(protocol.SessionConfirmData{Approved: false}) session.pushCommand(ctx, protocol.SessionCommand{Type: "session.confirm", Data: data}) case "event.user_question": - uq := decodeJobEvent[protocol.EventUserQuestion](ev.Data) + uq := decodeUnattendedEvent[protocol.EventUserQuestion](ev.Data) answer := "" if len(uq.RichOptions) > 0 { answer = uq.RichOptions[0].Title diff --git a/internal/daemon/job_runner.go b/internal/daemon/job_runner.go index 1e37ad4..86a1c7e 100644 --- a/internal/daemon/job_runner.go +++ b/internal/daemon/job_runner.go @@ -4,12 +4,10 @@ import ( "context" "encoding/json" "fmt" - "path/filepath" "regexp" "strings" "time" - "github.com/get-vix/vix/internal/config" "github.com/get-vix/vix/internal/daemon/jobs" "github.com/get-vix/vix/internal/protocol" ) @@ -39,163 +37,58 @@ func (s *Server) runJob(ctx context.Context, spec jobs.Spec, resolvedPrompt stri if runID == "" { runID = generateSessionID() } - session := NewSession(runID, s, nil, s.model, spec.CWD, "", false, - spec.AutoWrite(), spec.AutoDirs(), true /*headless*/, ctx) - session.origin = "vix" - session.trigger = &protocol.TriggerInfo{Type: spec.Trigger.Type, Ref: spec.ID} - session.title = jobRunTitle(spec, time.Now()) - // Expose the job's own directory (~/.vix/jobs/) to workflow templates as - // $(workflow.dir), so a run can persist state (e.g. a memory file) alongside - // its spec. Empty when the home directory is unavailable. Also mark it - // allowed so the run can read/write there even when the jobs directory lives - // outside $HOME and cwd (e.g. under a --config-dir override) — this flows to - // the file-tool path checks and the bash sandbox's writable set alike. - if jobsRoot := config.NewVixPaths("", s.homeVixDir, "").Jobs(); jobsRoot != "" { - session.jobDir = filepath.Join(jobsRoot, spec.ID) - session.addAllowedDir(session.jobDir) - } - - // Register so the web UI and session.list see the run while it's live. - s.sessionMu.Lock() - s.sessions[runID] = session - s.sessionMu.Unlock() - s.broadcastSessionsChanged() - defer func() { - s.sessionMu.Lock() - delete(s.sessions, runID) - s.sessionMu.Unlock() - session.cancel() - s.broadcastSessionsChanged() - }() - go session.Run() - - // Dispatch exactly like headless, resolving the prompt as $(workflow.prompt) - // when a workflow is involved: - // - inline workflow → session.workflow carrying the definition (the - // session registers it transiently and runs it); - // - named workflow_id → session.workflow by name; - // - neither → plain chat turn. - var startCmd protocol.SessionCommand - switch { - case spec.Workflow != nil: - raw, _ := json.Marshal(spec.Workflow) - data, _ := json.Marshal(protocol.SessionWorkflowData{Name: spec.Workflow.Name, Text: resolvedPrompt, Workflow: raw}) - startCmd = protocol.SessionCommand{Type: "session.workflow", Data: data} - case spec.WorkflowID != "": - data, _ := json.Marshal(protocol.SessionWorkflowData{Name: spec.WorkflowID, Text: resolvedPrompt}) - startCmd = protocol.SessionCommand{Type: "session.workflow", Data: data} - default: - data, _ := json.Marshal(protocol.SessionInputData{Text: resolvedPrompt}) - startCmd = protocol.SessionCommand{Type: "session.input", Data: data} + req := unattendedRunRequest{ + RunID: runID, + Model: s.model, + CWD: spec.CWD, + Title: jobRunTitle(spec, time.Now()), + Trigger: &protocol.TriggerInfo{Type: spec.Trigger.Type, Ref: spec.ID}, + Prompt: resolvedPrompt, + AutoWrite: spec.AutoWrite(), + AutoDirs: spec.AutoDirs(), + JobID: spec.ID, + SuppressFinishBroadcast: true, } - if !session.pushCommand(ctx, startCmd) { - return jobs.RunResult{ - Status: jobs.StatusError, - Err: "session refused start command", - Errors: []jobs.RunError{{Source: "start_refused", Message: "session refused start command"}}, - } - } - - // Consume the event stream (mandatory: emit blocks once eventChan fills - // with no reader), answering interactive events with unattended policy: - // confirmations are denied and recorded, questions take the first option, - // plans are approved — mirroring headless except for the deny. - var ( - finalText strings.Builder - agentTurns int - hadError bool - errMsg string - denials []string - ) - -consume: - for { - select { - case ev := <-session.eventChan: - switch ev.Type { - case "event.stream_chunk": - finalText.WriteString(decodeJobEvent[protocol.EventStreamChunk](ev.Data).Text) - case "event.stream_done": - agentTurns++ - case "event.confirm_request": - cr := decodeJobEvent[protocol.EventConfirmRequest](ev.Data) - denials = append(denials, cr.ToolName) - data, _ := json.Marshal(protocol.SessionConfirmData{Approved: false}) - session.pushCommand(ctx, protocol.SessionCommand{Type: "session.confirm", Data: data}) - case "event.user_question": - uq := decodeJobEvent[protocol.EventUserQuestion](ev.Data) - answer := "" - if len(uq.RichOptions) > 0 { - answer = uq.RichOptions[0].Title - } else if len(uq.Options) > 0 { - answer = uq.Options[0] - } - data, _ := json.Marshal(protocol.SessionUserAnswerData{Answer: answer}) - session.pushCommand(ctx, protocol.SessionCommand{Type: "session.user_answer", Data: data}) - case "event.plan_proposed": - data, _ := json.Marshal(protocol.SessionPlanActionData{Action: "approve"}) - session.pushCommand(ctx, protocol.SessionCommand{Type: "session.plan_action", Data: data}) - case "event.error": - hadError = true - errMsg = decodeJobEvent[protocol.EventError](ev.Data).Message - case "event.agent_done": - break consume - } - case <-ctx.Done(): - // Timeout or daemon shutdown: the session ctx (derived from ctx) - // is collapsing; persist what we have and report. - session.persist() - return jobs.RunResult{ - Status: jobs.StatusTimeout, - Err: "run cancelled: " + ctx.Err().Error(), - SessionID: runID, - AgentTurns: agentTurns, - Errors: []jobs.RunError{{Source: "timeout", Message: "run cancelled: " + ctx.Err().Error()}}, - } - case <-session.ctx.Done(): - break consume - } - } - - res := jobs.RunResult{Status: jobs.StatusOK, SessionID: runID, AgentTurns: agentTurns, Denials: denials} - if hadError { - res.Status = jobs.StatusError - res.Err = errMsg - res.Errors = append(res.Errors, jobs.RunError{Source: "agent", Message: errMsg}) - } - if len(denials) > 0 && res.Err == "" { - res.Err = "needed approval for: " + strings.Join(denials, "; ") - } - if len(denials) > 0 { - res.Errors = append(res.Errors, jobs.RunError{Source: "denied", Message: "needed approval for: " + strings.Join(denials, "; ")}) + if spec.Workflow != nil { + req.Workflow.Inline = spec.Workflow + } else if spec.WorkflowID != "" { + req.Workflow.Name = spec.WorkflowID } + run := s.runUnattendedSession(ctx, req, jobUnattendedPolicy) + res := jobRunResultFromUnattended(run) // Skip rules — a skipped run leaves no trace: // cheap-poll: no agent step executed (a poll workflow whose execute_if // gate didn't pass — bash steps never call the LLM); // heartbeat OK: the model said nothing needs attention. - if res.Status == jobs.StatusOK && (agentTurns == 0 || isHeartbeatOK(finalText.String())) { - deleteSessionRecord(session.paths, runID) - return jobs.RunResult{Status: jobs.StatusSkipped, SessionID: runID, AgentTurns: agentTurns} + if res.Status == jobs.StatusOK && (run.AgentTurns == 0 || isHeartbeatOK(run.FinalText)) { + if run.session != nil { + deleteSessionRecord(run.session.paths, run.SessionID) + s.broadcastSessionsChanged() + } + return jobs.RunResult{Status: jobs.StatusSkipped, SessionID: run.SessionID, AgentTurns: run.AgentTurns} } // Every other finished run lands in open/: visible in the Vix-initiated // sessions group until the user dismisses it (or retention sweeps it). - session.jobStatus = res.Status - // Successful GitHub-plan runs open their findings with a deterministic - // header line naming the item they picked; turn that into a per-item session - // title (e.g. "[Plan GitHub issues (get-vix/vix)] Addressing issue #29 — …"). - // Other jobs (and the "nothing new"/error branches) keep the static title. - if res.Status == jobs.StatusOK { - if title, ok := issuePlanTitle(spec, finalText.String()); ok { - session.mu.Lock() - session.title = title - session.mu.Unlock() + if run.session != nil { + run.session.jobStatus = res.Status + // Successful GitHub-plan runs open their findings with a deterministic + // header line naming the item they picked; turn that into a per-item session + // title (e.g. "[Plan GitHub issues (get-vix/vix)] Addressing issue #29 — …"). + // Other jobs (and the "nothing new"/error branches) keep the static title. + if res.Status == jobs.StatusOK { + if title, ok := issuePlanTitle(spec, run.FinalText); ok { + run.session.mu.Lock() + run.session.title = title + run.session.mu.Unlock() + } } + run.session.persist() + sweepJobRunRecords(run.session.paths, spec.ID) + s.broadcastSessionsChanged() } - session.persist() - sweepJobRunRecords(session.paths, spec.ID) // Failures nobody saw get a synthetic explainer session on top of the run // record, so the next TUI launch surfaces them. @@ -205,6 +98,68 @@ consume: return res } +func jobUnattendedPolicy(ctx context.Context, session *Session, ev protocol.SessionEvent) (bool, error) { + switch ev.Type { + case "event.confirm_request": + data, _ := json.Marshal(protocol.SessionConfirmData{Approved: false}) + return session.pushCommand(ctx, protocol.SessionCommand{Type: "session.confirm", Data: data}), nil + case "event.user_question": + uq := decodeUnattendedEvent[protocol.EventUserQuestion](ev.Data) + answer := "" + if len(uq.RichOptions) > 0 { + answer = uq.RichOptions[0].Title + } else if len(uq.Options) > 0 { + answer = uq.Options[0] + } + data, _ := json.Marshal(protocol.SessionUserAnswerData{Answer: answer}) + return session.pushCommand(ctx, protocol.SessionCommand{Type: "session.user_answer", Data: data}), nil + case "event.plan_proposed": + data, _ := json.Marshal(protocol.SessionPlanActionData{Action: "approve"}) + return session.pushCommand(ctx, protocol.SessionCommand{Type: "session.plan_action", Data: data}), nil + default: + return false, nil + } +} + +func jobRunResultFromUnattended(run unattendedRunResult) jobs.RunResult { + if run.TimedOut { + return jobs.RunResult{ + Status: jobs.StatusTimeout, + Err: run.Err, + SessionID: run.SessionID, + AgentTurns: run.AgentTurns, + Errors: []jobs.RunError{{Source: "timeout", Message: run.Err}}, + } + } + + res := jobs.RunResult{ + Status: jobs.StatusOK, + SessionID: run.SessionID, + AgentTurns: run.AgentTurns, + Denials: run.ConfirmRequests, + } + if run.HadError { + errMsg := run.Err + if errMsg == "" { + errMsg = "agent run failed" + } + errSource := run.ErrSource + if errSource == "" { + errSource = "agent" + } + res.Status = jobs.StatusError + res.Err = errMsg + res.Errors = append(res.Errors, jobs.RunError{Source: errSource, Message: errMsg}) + } + if len(run.ConfirmRequests) > 0 && res.Err == "" { + res.Err = "needed approval for: " + strings.Join(run.ConfirmRequests, "; ") + } + if len(run.ConfirmRequests) > 0 { + res.Errors = append(res.Errors, jobs.RunError{Source: "denied", Message: "needed approval for: " + strings.Join(run.ConfirmRequests, "; ")}) + } + return res +} + // jobRunTitle builds the display title of a job-run session, e.g. // "Heartbeat - 06/12/2026 9:30 AM". func jobRunTitle(spec jobs.Spec, t time.Time) string { @@ -322,14 +277,3 @@ func isHeartbeatOK(text string) bool { rest := strings.TrimSpace(strings.TrimSuffix(strings.TrimPrefix(t, heartbeatOKToken), heartbeatOKToken)) return len(rest) <= heartbeatOKSlop } - -// decodeJobEvent unmarshals an event payload into the given type. -func decodeJobEvent[T any](data any) T { - var out T - raw, err := json.Marshal(data) - if err != nil { - return out - } - json.Unmarshal(raw, &out) - return out -} diff --git a/internal/daemon/job_runner_test.go b/internal/daemon/job_runner_test.go new file mode 100644 index 0000000..881da73 --- /dev/null +++ b/internal/daemon/job_runner_test.go @@ -0,0 +1,117 @@ +package daemon + +import ( + "context" + "errors" + "reflect" + "testing" + + "github.com/get-vix/vix/internal/daemon/jobs" +) + +func TestJobRunResultFromUnattendedTimeout(t *testing.T) { + res := jobRunResultFromUnattended(unattendedRunResult{ + SessionID: "run-1", + AgentTurns: 2, + TimedOut: true, + Err: "run cancelled: context deadline exceeded", + }) + + if res.Status != jobs.StatusTimeout { + t.Fatalf("status = %q, want %q", res.Status, jobs.StatusTimeout) + } + if res.Err != "run cancelled: context deadline exceeded" { + t.Fatalf("err = %q", res.Err) + } + if res.SessionID != "run-1" || res.AgentTurns != 2 { + t.Fatalf("session/turns = %q/%d", res.SessionID, res.AgentTurns) + } + if len(res.Errors) != 1 || res.Errors[0].Source != "timeout" || res.Errors[0].Message != res.Err { + t.Fatalf("errors = %+v", res.Errors) + } +} + +func TestJobRunResultFromUnattendedDenials(t *testing.T) { + res := jobRunResultFromUnattended(unattendedRunResult{ + SessionID: "run-2", + AgentTurns: 1, + ConfirmRequests: []string{"bash", "edit_file"}, + }) + + if res.Status != jobs.StatusOK { + t.Fatalf("status = %q, want %q", res.Status, jobs.StatusOK) + } + wantErr := "needed approval for: bash; edit_file" + if res.Err != wantErr { + t.Fatalf("err = %q, want %q", res.Err, wantErr) + } + if !reflect.DeepEqual(res.Denials, []string{"bash", "edit_file"}) { + t.Fatalf("denials = %+v", res.Denials) + } + if len(res.Errors) != 1 || res.Errors[0].Source != "denied" || res.Errors[0].Message != wantErr { + t.Fatalf("errors = %+v", res.Errors) + } +} + +func TestJobRunResultFromUnattendedAgentErrorWithDenials(t *testing.T) { + res := jobRunResultFromUnattended(unattendedRunResult{ + SessionID: "run-3", + ConfirmRequests: []string{"bash"}, + HadError: true, + ErrSource: "agent", + Err: "model failed", + }) + + if res.Status != jobs.StatusError { + t.Fatalf("status = %q, want %q", res.Status, jobs.StatusError) + } + if res.Err != "model failed" { + t.Fatalf("err = %q", res.Err) + } + if len(res.Errors) != 2 { + t.Fatalf("errors = %+v", res.Errors) + } + if res.Errors[0].Source != "agent" || res.Errors[0].Message != "model failed" { + t.Fatalf("first error = %+v", res.Errors[0]) + } + if res.Errors[1].Source != "denied" || res.Errors[1].Message != "needed approval for: bash" { + t.Fatalf("second error = %+v", res.Errors[1]) + } +} + +func TestJobRunResultFromUnattendedEmptyErrorStillFails(t *testing.T) { + res := jobRunResultFromUnattended(unattendedRunResult{ + SessionID: "run-4", + HadError: true, + }) + + if res.Status != jobs.StatusError { + t.Fatalf("status = %q, want %q", res.Status, jobs.StatusError) + } + if res.Err != "agent run failed" { + t.Fatalf("err = %q, want fallback", res.Err) + } + if len(res.Errors) != 1 || res.Errors[0].Source != "agent" || res.Errors[0].Message != "agent run failed" { + t.Fatalf("errors = %+v", res.Errors) + } +} + +func TestUnattendedPolicyErrorAfterCancelMapsToTimeout(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + res := unattendedPolicyErrorResult(ctx, unattendedRunResult{SessionID: "run-5"}, "partial output", errors.New("unhandled unattended event: event.confirm_request")) + + if !res.TimedOut { + t.Fatal("TimedOut = false, want true") + } + if res.ErrSource != "timeout" { + t.Fatalf("ErrSource = %q, want timeout", res.ErrSource) + } + if res.Err != "run cancelled: context canceled" { + t.Fatalf("Err = %q", res.Err) + } + if res.FinalText != "partial output" { + t.Fatalf("FinalText = %q", res.FinalText) + } +} diff --git a/internal/daemon/remote_control.go b/internal/daemon/remote_control.go new file mode 100644 index 0000000..f69da7d --- /dev/null +++ b/internal/daemon/remote_control.go @@ -0,0 +1,263 @@ +package daemon + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "time" + + "github.com/get-vix/vix/internal/config" + "github.com/get-vix/vix/internal/protocol" +) + +// RemoteControlConfig configures opt-in remote control from chat services. +type RemoteControlConfig struct { + Enabled bool `json:"enabled"` + CWD string `json:"cwd"` + Workflow string `json:"workflow,omitempty"` + MaxConcurrentRuns int `json:"max_concurrent_runs,omitempty"` + Telegram TelegramRemoteConfig `json:"telegram,omitempty"` + WhatsApp WhatsAppRemoteConfig `json:"whatsapp,omitempty"` +} + +func (cfg RemoteControlConfig) maxConcurrentRuns() int { + if cfg.MaxConcurrentRuns <= 0 { + return 1 + } + return cfg.MaxConcurrentRuns +} + +type TelegramRemoteConfig struct { + Enabled bool `json:"enabled"` + BotToken string `json:"bot_token"` + AllowedChatIDs []string `json:"allowed_chat_ids"` + PollIntervalMs int `json:"poll_interval_ms,omitempty"` +} + +type WhatsAppRemoteConfig struct { + Enabled bool `json:"enabled"` + AccessToken string `json:"access_token"` + AppSecret string `json:"app_secret"` + PhoneNumberID string `json:"phone_number_id"` + VerifyToken string `json:"verify_token"` + GraphAPIVersion string `json:"graph_api_version,omitempty"` + WebhookAddr string `json:"webhook_addr,omitempty"` + AllowedContacts []string `json:"allowed_contacts"` +} + +type remoteReplyFunc func(ctx context.Context, text string) error + +type remoteMessage struct { + Provider string + SenderID string + Text string + Reply remoteReplyFunc +} + +type remoteHTTPClient interface { + Do(*http.Request) (*http.Response, error) +} + +const remoteControlBusyMessage = "vix remote control is busy; try again after the current run finishes." + +type remoteControl struct { + server *Server + cfg RemoteControlConfig + http remoteHTTPClient + runs chan struct{} + runPrompt func(context.Context, string, string, string) (string, error) +} + +func newRemoteControl(server *Server, cfg RemoteControlConfig, hc remoteHTTPClient) *remoteControl { + if hc == nil { + hc = http.DefaultClient + } + rc := &remoteControl{ + server: server, + cfg: cfg, + http: hc, + runs: make(chan struct{}, cfg.maxConcurrentRuns()), + } + if server != nil { + rc.runPrompt = server.runRemotePrompt + } + return rc +} + +func LoadRemoteControlConfig() (RemoteControlConfig, error) { + p := filepath.Join(config.HomeVixDir(), "settings.json") + data, err := os.ReadFile(p) + if err != nil { + if os.IsNotExist(err) { + return RemoteControlConfig{}, nil + } + return RemoteControlConfig{}, err + } + var cfg struct { + RemoteControl RemoteControlConfig `json:"remote_control"` + } + if err := json.Unmarshal(data, &cfg); err != nil { + return RemoteControlConfig{}, err + } + return cfg.RemoteControl, nil +} + +func (s *Server) StartRemoteControl(ctx context.Context, cfg RemoteControlConfig) error { + if !cfg.Enabled { + return nil + } + if strings.TrimSpace(cfg.CWD) == "" { + return fmt.Errorf("remote control: missing cwd") + } + rc := newRemoteControl(s, cfg, nil) + started := false + if cfg.Telegram.Enabled { + if err := rc.startTelegram(ctx); err != nil { + return err + } + started = true + } + if cfg.WhatsApp.Enabled { + if err := rc.startWhatsApp(ctx); err != nil { + return err + } + started = true + } + if !started { + return fmt.Errorf("remote control: enabled but no provider is enabled") + } + return nil +} + +func (rc *remoteControl) handleMessage(ctx context.Context, msg remoteMessage) { + text := strings.TrimSpace(msg.Text) + if text == "" { + return + } + if !rc.tryAcquireRun() { + if err := msg.Reply(ctx, remoteControlBusyMessage); err != nil { + LogError("remote control: reply to %s %s failed: %v", msg.Provider, msg.SenderID, err) + } + return + } + + LogInfo("remote control: received %s message from %s", msg.Provider, msg.SenderID) + var result string + var err error + func() { + defer rc.releaseRun() + result, err = rc.runPrompt(ctx, rc.cfg.CWD, rc.cfg.Workflow, text) + }() + if err != nil { + result = "vix remote control error: " + err.Error() + LogError("remote control: %s", err) + } + if err := msg.Reply(ctx, result); err != nil { + LogError("remote control: reply to %s %s failed: %v", msg.Provider, msg.SenderID, err) + } +} + +func (rc *remoteControl) tryAcquireRun() bool { + select { + case rc.runs <- struct{}{}: + return true + default: + return false + } +} + +func (rc *remoteControl) releaseRun() { + select { + case <-rc.runs: + default: + } +} + +func (s *Server) runRemotePrompt(ctx context.Context, cwd, workflow, prompt string) (string, error) { + req := unattendedRunRequest{ + Model: s.model, + CWD: cwd, + Title: "Remote control - " + time.Now().Format(jobTitleTimeFormat), + Trigger: &protocol.TriggerInfo{Type: "remote_control", Ref: "remote"}, + Prompt: prompt, + AutoWrite: false, + AutoDirs: false, + } + if workflow != "" { + req.Workflow.Name = workflow + } + res := s.runUnattendedSession(ctx, req, remoteUnattendedPolicy) + return remotePromptResultFromUnattended(res) +} + +func remotePromptResultFromUnattended(res unattendedRunResult) (string, error) { + if res.HadError { + if strings.TrimSpace(res.FinalText) != "" && res.ErrSource == "agent" { + return res.FinalText, nil + } + if strings.TrimSpace(res.Err) == "" { + return res.FinalText, errors.New("remote control run failed") + } + return res.FinalText, errors.New(res.Err) + } + return res.FinalText, nil +} + +func remoteUnattendedPolicy(ctx context.Context, session *Session, ev protocol.SessionEvent) (bool, error) { + cmd, err := remoteUnattendedPolicyCommand(ev) + if err != nil { + return true, err + } + return session.pushCommand(ctx, cmd), nil +} + +func remoteUnattendedPolicyCommand(ev protocol.SessionEvent) (protocol.SessionCommand, error) { + switch ev.Type { + case "event.confirm_request": + data, _ := json.Marshal(protocol.SessionConfirmData{Approved: false}) + return protocol.SessionCommand{Type: "session.confirm", Data: data}, nil + case "event.user_question": + return protocol.SessionCommand{}, fmt.Errorf("remote control requires an interactive answer") + case "event.plan_proposed": + return protocol.SessionCommand{}, fmt.Errorf("remote control requires interactive approval") + default: + return protocol.SessionCommand{}, fmt.Errorf("unsupported unattended event: %s", ev.Type) + } +} + +func authorizedRemoteID(id string, allowed []string) bool { + if len(allowed) == 0 { + return false + } + for _, v := range allowed { + if strings.TrimSpace(v) == id { + return true + } + } + return false +} + +func postForm(ctx context.Context, hc remoteHTTPClient, endpoint string, form url.Values) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, strings.NewReader(form.Encode())) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + resp, err := hc.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("remote provider returned %s: %s", resp.Status, strings.TrimSpace(string(b))) + } + return nil +} diff --git a/internal/daemon/remote_control_telegram.go b/internal/daemon/remote_control_telegram.go new file mode 100644 index 0000000..b8348d6 --- /dev/null +++ b/internal/daemon/remote_control_telegram.go @@ -0,0 +1,129 @@ +package daemon + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +const telegramAPIBase = "https://api.telegram.org" + +type telegramUpdatesResponse struct { + OK bool `json:"ok"` + Result []telegramUpdate `json:"result"` +} + +type telegramUpdate struct { + UpdateID int64 `json:"update_id"` + Message telegramMessage `json:"message"` +} + +type telegramMessage struct { + Text string `json:"text"` + Chat telegramChat `json:"chat"` +} + +type telegramChat struct { + ID int64 `json:"id"` +} + +func (rc *remoteControl) startTelegram(ctx context.Context) error { + cfg := rc.cfg.Telegram + if strings.TrimSpace(cfg.BotToken) == "" { + return fmt.Errorf("remote control telegram: missing bot_token") + } + if len(cfg.AllowedChatIDs) == 0 { + return fmt.Errorf("remote control telegram: missing allowed_chat_ids") + } + interval := time.Duration(cfg.PollIntervalMs) * time.Millisecond + if interval <= 0 { + interval = 2 * time.Second + } + go rc.pollTelegram(ctx, cfg, interval) + LogInfo("remote control: telegram polling enabled") + return nil +} + +func (rc *remoteControl) pollTelegram(ctx context.Context, cfg TelegramRemoteConfig, interval time.Duration) { + var offset int64 + for { + updates, err := rc.getTelegramUpdates(ctx, cfg.BotToken, offset) + if err != nil { + LogError("remote control telegram: getUpdates failed: %v", err) + } else { + for _, upd := range updates { + if upd.UpdateID >= offset { + offset = upd.UpdateID + 1 + } + chatID := strconv.FormatInt(upd.Message.Chat.ID, 10) + if !authorizedRemoteID(chatID, cfg.AllowedChatIDs) || strings.TrimSpace(upd.Message.Text) == "" { + continue + } + msg := remoteMessage{ + Provider: "telegram", + SenderID: chatID, + Text: upd.Message.Text, + Reply: func(replyCtx context.Context, text string) error { + return rc.sendTelegramMessage(replyCtx, cfg.BotToken, chatID, text) + }, + } + go rc.handleMessage(ctx, msg) + } + } + + select { + case <-ctx.Done(): + return + case <-time.After(interval): + } + } +} + +func (rc *remoteControl) getTelegramUpdates(ctx context.Context, token string, offset int64) ([]telegramUpdate, error) { + endpoint := telegramAPIBase + "/bot" + token + "/getUpdates" + q := url.Values{} + q.Set("timeout", "20") + if offset > 0 { + q.Set("offset", strconv.FormatInt(offset, 10)) + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint+"?"+q.Encode(), nil) + if err != nil { + return nil, err + } + resp, err := rc.http.Do(req) + if err != nil { + return nil, redactTelegramError(err, token) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("telegram getUpdates returned %s", resp.Status) + } + var out telegramUpdatesResponse + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, err + } + if !out.OK { + return nil, fmt.Errorf("telegram getUpdates returned ok=false") + } + return out.Result, nil +} + +func (rc *remoteControl) sendTelegramMessage(ctx context.Context, token, chatID, text string) error { + endpoint := telegramAPIBase + "/bot" + token + "/sendMessage" + if err := postForm(ctx, rc.http, endpoint, url.Values{"chat_id": {chatID}, "text": {text}}); err != nil { + return redactTelegramError(err, token) + } + return nil +} + +func redactTelegramError(err error, token string) error { + if err == nil || token == "" { + return err + } + return fmt.Errorf("%s", strings.ReplaceAll(err.Error(), token, "[REDACTED]")) +} diff --git a/internal/daemon/remote_control_test.go b/internal/daemon/remote_control_test.go new file mode 100644 index 0000000..56bdffc --- /dev/null +++ b/internal/daemon/remote_control_test.go @@ -0,0 +1,292 @@ +package daemon + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" + + "github.com/get-vix/vix/internal/protocol" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) Do(req *http.Request) (*http.Response, error) { return f(req) } + +func TestAuthorizedRemoteIDRequiresAllowlist(t *testing.T) { + if authorizedRemoteID("123", nil) { + t.Fatal("empty allowlist must deny") + } + if authorizedRemoteID("123", []string{"456"}) { + t.Fatal("unlisted sender must deny") + } + if !authorizedRemoteID("123", []string{" 123 "}) { + t.Fatal("listed sender should be authorized") + } +} + +func TestRemoteControlDefaults(t *testing.T) { + cfg := RemoteControlConfig{} + if got := cfg.maxConcurrentRuns(); got != 1 { + t.Fatalf("maxConcurrentRuns() = %d, want 1", got) + } + if got := cfg.WhatsApp.graphAPIVersion(); got != defaultWhatsAppGraphAPIVersion { + t.Fatalf("graphAPIVersion() = %q, want %q", got, defaultWhatsAppGraphAPIVersion) + } +} + +func TestRemoteControlConfigHonorsOptionalFields(t *testing.T) { + cfg := RemoteControlConfig{MaxConcurrentRuns: 3, WhatsApp: WhatsAppRemoteConfig{GraphAPIVersion: "v21.0"}} + if got := cfg.maxConcurrentRuns(); got != 3 { + t.Fatalf("maxConcurrentRuns() = %d, want 3", got) + } + if got := cfg.WhatsApp.graphAPIVersion(); got != "v21.0" { + t.Fatalf("graphAPIVersion() = %q, want v21.0", got) + } + if got := (WhatsAppRemoteConfig{GraphAPIVersion: " v21.0 "}).graphAPIVersion(); got != "v21.0" { + t.Fatalf("graphAPIVersion() = %q, want v21.0", got) + } +} + +func TestRemoteUnattendedPolicyDoesNotAutoApprove(t *testing.T) { + cmd, err := remoteUnattendedPolicyCommand(protocol.SessionEvent{Type: "event.confirm_request"}) + if err != nil { + t.Fatalf("confirm request handling: %v", err) + } + if cmd.Type != "session.confirm" || !strings.Contains(string(cmd.Data), `"approved":false`) { + t.Fatalf("confirm request command = %+v data=%s, want denial", cmd, string(cmd.Data)) + } + + _, err = remoteUnattendedPolicyCommand(protocol.SessionEvent{Type: "event.plan_proposed"}) + if err == nil || !strings.Contains(err.Error(), "requires interactive approval") { + t.Fatalf("plan proposal err = %v, want interactive approval error", err) + } + + _, err = remoteUnattendedPolicyCommand(protocol.SessionEvent{Type: "event.user_question"}) + if err == nil || !strings.Contains(err.Error(), "requires an interactive answer") { + t.Fatalf("user question err = %v, want interactive answer error", err) + } +} + +func TestRemotePromptResultKeepsFinalTextWhenAgentErrorHasOutput(t *testing.T) { + text, err := remotePromptResultFromUnattended(unattendedRunResult{ + FinalText: "partial answer", + HadError: true, + Err: "late agent error", + ErrSource: "agent", + }) + if err != nil { + t.Fatalf("remotePromptResultFromUnattended err = %v, want nil", err) + } + if text != "partial answer" { + t.Fatalf("remotePromptResultFromUnattended text = %q, want partial answer", text) + } +} + +func TestRemoteControlRejectsMessageWhenConcurrencyLimitReached(t *testing.T) { + rc := newRemoteControl(nil, RemoteControlConfig{MaxConcurrentRuns: 1}, nil) + if !rc.tryAcquireRun() { + t.Fatal("tryAcquireRun() = false, want initial slot") + } + defer rc.releaseRun() + + var replies []string + rc.handleMessage(context.Background(), remoteMessage{ + Provider: "telegram", + SenderID: "42", + Text: "run tests", + Reply: func(_ context.Context, text string) error { + replies = append(replies, text) + return nil + }, + }) + + if len(replies) != 1 { + t.Fatalf("replies = %d, want 1", len(replies)) + } + if replies[0] != remoteControlBusyMessage { + t.Fatalf("reply = %q, want %q", replies[0], remoteControlBusyMessage) + } +} + +func TestRemoteControlReleasesRunSlotBeforeReplyReturns(t *testing.T) { + rc := newRemoteControl(nil, RemoteControlConfig{MaxConcurrentRuns: 1}, nil) + rc.runPrompt = func(context.Context, string, string, string) (string, error) { + return "done", nil + } + + replyStarted := make(chan struct{}) + releaseReply := make(chan struct{}) + done := make(chan struct{}) + go func() { + rc.handleMessage(context.Background(), remoteMessage{ + Provider: "telegram", + SenderID: "42", + Text: "run tests", + Reply: func(_ context.Context, text string) error { + close(replyStarted) + <-releaseReply + return nil + }, + }) + close(done) + }() + + select { + case <-replyStarted: + case <-time.After(time.Second): + t.Fatal("reply did not start") + } + if !rc.tryAcquireRun() { + close(releaseReply) + <-done + t.Fatal("run slot remained occupied while reply was blocked") + } + rc.releaseRun() + close(releaseReply) + <-done +} + +func TestTelegramSendMessageRedactsBotTokenFromErrors(t *testing.T) { + secret := "123456:secret-token" + rc := &remoteControl{http: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return nil, &url.Error{Op: "Post", URL: req.URL.String(), Err: fmt.Errorf("dial failed")} + })} + + err := rc.sendTelegramMessage(context.Background(), secret, "42", "hello") + if err == nil { + t.Fatal("sendTelegramMessage error = nil, want redacted error") + } + if strings.Contains(err.Error(), secret) { + t.Fatalf("error leaked bot token: %v", err) + } +} + +func TestWhatsAppStartFailsWhenWebhookAddrAlreadyBound(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + + rc := &remoteControl{cfg: RemoteControlConfig{WhatsApp: WhatsAppRemoteConfig{ + Enabled: true, + AccessToken: "access", + AppSecret: "secret", + PhoneNumberID: "phone", + VerifyToken: "verify", + WebhookAddr: ln.Addr().String(), + AllowedContacts: []string{"15551234567"}, + }}} + if err := rc.startWhatsApp(context.Background()); err == nil { + t.Fatal("startWhatsApp error = nil, want bind failure") + } +} + +func TestTelegramSendMessagePostsChatIDAndText(t *testing.T) { + var gotPath, gotBody string + rc := &remoteControl{http: roundTripFunc(func(req *http.Request) (*http.Response, error) { + gotPath = req.URL.Path + b, _ := io.ReadAll(req.Body) + gotBody = string(b) + return &http.Response{StatusCode: http.StatusOK, Status: "200 OK", Body: io.NopCloser(strings.NewReader(`{"ok":true}`))}, nil + })} + + if err := rc.sendTelegramMessage(context.Background(), "token", "42", "hello vix"); err != nil { + t.Fatalf("sendTelegramMessage: %v", err) + } + if gotPath != "/bottoken/sendMessage" { + t.Fatalf("path = %q", gotPath) + } + if !strings.Contains(gotBody, "chat_id=42") || !strings.Contains(gotBody, "text=hello+vix") { + t.Fatalf("unexpected body %q", gotBody) + } +} + +func TestWhatsAppSendMessageUsesConfiguredGraphAPIVersion(t *testing.T) { + var gotURL string + rc := &remoteControl{http: roundTripFunc(func(req *http.Request) (*http.Response, error) { + gotURL = req.URL.String() + return &http.Response{StatusCode: http.StatusOK, Status: "200 OK", Body: io.NopCloser(strings.NewReader(`{}`))}, nil + })} + cfg := WhatsAppRemoteConfig{ + AccessToken: "access", + PhoneNumberID: "phone", + GraphAPIVersion: "v21.0", + } + + if err := rc.sendWhatsAppMessage(context.Background(), cfg, "15551234567", "hello"); err != nil { + t.Fatalf("sendWhatsAppMessage: %v", err) + } + if gotURL != "https://graph.facebook.com/v21.0/phone/messages" { + t.Fatalf("url = %q", gotURL) + } +} + +func TestTelegramGetUpdatesRedactsBotTokenFromErrors(t *testing.T) { + secret := "123456:secret-token" + rc := &remoteControl{http: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return nil, &url.Error{Op: "Get", URL: req.URL.String(), Err: fmt.Errorf("dial failed")} + })} + + _, err := rc.getTelegramUpdates(context.Background(), secret, 0) + if err == nil { + t.Fatal("getTelegramUpdates error = nil, want redacted error") + } + if strings.Contains(err.Error(), secret) { + t.Fatalf("error leaked bot token: %v", err) + } +} + +func TestWhatsAppWebhookVerificationAndAllowlist(t *testing.T) { + handled := make(chan remoteMessage, 1) + rc := &remoteControl{} + cfg := WhatsAppRemoteConfig{VerifyToken: "verify", AppSecret: "secret", AllowedContacts: []string{"15551234567"}} + h := rc.whatsAppWebhookHandlerWithHandle(context.Background(), cfg, func(_ context.Context, msg remoteMessage) { + handled <- msg + }) + + req := httptest.NewRequest(http.MethodGet, "/whatsapp?hub.mode=subscribe&hub.verify_token=verify&hub.challenge=abc", nil) + w := httptest.NewRecorder() + h(w, req) + if w.Code != http.StatusOK || w.Body.String() != "abc" { + t.Fatalf("verification response = %d %q", w.Code, w.Body.String()) + } + + body := `{"entry":[{"changes":[{"value":{"messages":[{"from":"15551234567","type":"text","text":{"body":"run tests"}},{"from":"blocked","type":"text","text":{"body":"ignore"}}]}}]}]}` + req = httptest.NewRequest(http.MethodPost, "/whatsapp", strings.NewReader(body)) + req.Header.Set("X-Hub-Signature-256", whatsappTestSignature(body, "secret")) + w = httptest.NewRecorder() + h(w, req) + if w.Code != http.StatusOK { + t.Fatalf("post response = %d", w.Code) + } + select { + case msg := <-handled: + if msg.SenderID != "15551234567" || msg.Text != "run tests" { + t.Fatalf("unexpected message: %+v", msg) + } + case <-time.After(time.Second): + t.Fatal("handled 0 messages, want 1") + } + select { + case msg := <-handled: + t.Fatalf("unexpected extra message: %+v", msg) + default: + } +} + +func whatsappTestSignature(body, secret string) string { + mac := hmac.New(sha256.New, []byte(secret)) + _, _ = mac.Write([]byte(body)) + return "sha256=" + hex.EncodeToString(mac.Sum(nil)) +} diff --git a/internal/daemon/remote_control_whatsapp.go b/internal/daemon/remote_control_whatsapp.go new file mode 100644 index 0000000..fb45091 --- /dev/null +++ b/internal/daemon/remote_control_whatsapp.go @@ -0,0 +1,189 @@ +package daemon + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "time" +) + +const defaultWhatsAppGraphAPIVersion = "v20.0" + +type whatsappWebhookPayload struct { + Entry []struct { + Changes []struct { + Value struct { + Messages []struct { + From string `json:"from"` + Text struct { + Body string `json:"body"` + } `json:"text"` + Type string `json:"type"` + } `json:"messages"` + } `json:"value"` + } `json:"changes"` + } `json:"entry"` +} + +func (cfg WhatsAppRemoteConfig) graphAPIVersion() string { + v := strings.TrimSpace(cfg.GraphAPIVersion) + if v == "" { + return defaultWhatsAppGraphAPIVersion + } + return v +} + +func (rc *remoteControl) startWhatsApp(ctx context.Context) error { + cfg := rc.cfg.WhatsApp + if strings.TrimSpace(cfg.AccessToken) == "" { + return fmt.Errorf("remote control whatsapp: missing access_token") + } + if strings.TrimSpace(cfg.AppSecret) == "" { + return fmt.Errorf("remote control whatsapp: missing app_secret") + } + if strings.TrimSpace(cfg.PhoneNumberID) == "" { + return fmt.Errorf("remote control whatsapp: missing phone_number_id") + } + if strings.TrimSpace(cfg.VerifyToken) == "" { + return fmt.Errorf("remote control whatsapp: missing verify_token") + } + if len(cfg.AllowedContacts) == 0 { + return fmt.Errorf("remote control whatsapp: missing allowed_contacts") + } + addr := strings.TrimSpace(cfg.WebhookAddr) + if addr == "" { + addr = "127.0.0.1:1340" + } + ln, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("remote control whatsapp: listen %s: %w", addr, err) + } + mux := http.NewServeMux() + mux.HandleFunc("/whatsapp", rc.whatsAppWebhookHandler(ctx, cfg)) + server := &http.Server{Handler: mux} + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = server.Shutdown(shutdownCtx) + }() + go func() { + LogInfo("remote control: whatsapp webhook listening on %s", addr) + if err := server.Serve(ln); err != nil && err != http.ErrServerClosed { + LogError("remote control whatsapp: webhook server failed: %v", err) + } + }() + return nil +} + +func (rc *remoteControl) whatsAppWebhookHandler(ctx context.Context, cfg WhatsAppRemoteConfig) http.HandlerFunc { + return rc.whatsAppWebhookHandlerWithHandle(ctx, cfg, rc.handleMessage) +} + +func (rc *remoteControl) whatsAppWebhookHandlerWithHandle(ctx context.Context, cfg WhatsAppRemoteConfig, handle func(context.Context, remoteMessage)) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + mode := r.URL.Query().Get("hub.mode") + verifyToken := r.URL.Query().Get("hub.verify_token") + challenge := r.URL.Query().Get("hub.challenge") + if mode == "subscribe" && verifyToken == cfg.VerifyToken { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(challenge)) + return + } + http.Error(w, "forbidden", http.StatusForbidden) + case http.MethodPost: + defer r.Body.Close() + body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) + if err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + if !validWhatsAppSignature(body, r.Header.Get("X-Hub-Signature-256"), cfg.AppSecret) { + http.Error(w, "forbidden", http.StatusForbidden) + return + } + var payload whatsappWebhookPayload + if err := json.Unmarshal(body, &payload); err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusOK) + for _, entry := range payload.Entry { + for _, change := range entry.Changes { + for _, wm := range change.Value.Messages { + if wm.Type != "text" || !authorizedRemoteID(wm.From, cfg.AllowedContacts) || strings.TrimSpace(wm.Text.Body) == "" { + continue + } + from := wm.From + msg := remoteMessage{ + Provider: "whatsapp", + SenderID: from, + Text: wm.Text.Body, + Reply: func(replyCtx context.Context, text string) error { + return rc.sendWhatsAppMessage(replyCtx, cfg, from, text) + }, + } + go handle(ctx, msg) + } + } + } + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } + } +} + +func validWhatsAppSignature(body []byte, signature, appSecret string) bool { + if !strings.HasPrefix(signature, "sha256=") || appSecret == "" { + return false + } + got, err := hex.DecodeString(strings.TrimPrefix(signature, "sha256=")) + if err != nil { + return false + } + mac := hmac.New(sha256.New, []byte(appSecret)) + _, _ = mac.Write(body) + want := mac.Sum(nil) + return hmac.Equal(got, want) +} + +func (rc *remoteControl) sendWhatsAppMessage(ctx context.Context, cfg WhatsAppRemoteConfig, to, text string) error { + endpoint := fmt.Sprintf("https://graph.facebook.com/%s/%s/messages", cfg.graphAPIVersion(), cfg.PhoneNumberID) + payload := map[string]any{ + "messaging_product": "whatsapp", + "to": to, + "type": "text", + "text": map[string]string{ + "body": text, + }, + } + var body bytes.Buffer + if err := json.NewEncoder(&body).Encode(payload); err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, &body) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+cfg.AccessToken) + req.Header.Set("Content-Type", "application/json") + resp, err := rc.http.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("whatsapp send returned %s", resp.Status) + } + return nil +} diff --git a/internal/daemon/unattended_runner.go b/internal/daemon/unattended_runner.go new file mode 100644 index 0000000..d363137 --- /dev/null +++ b/internal/daemon/unattended_runner.go @@ -0,0 +1,243 @@ +package daemon + +import ( + "context" + "encoding/json" + "fmt" + "path/filepath" + "strings" + + "github.com/get-vix/vix/internal/config" + "github.com/get-vix/vix/internal/protocol" + wf "github.com/get-vix/vix/internal/workflow" +) + +type unattendedWorkflowRequest struct { + Name string + Inline *wf.Def +} + +type unattendedRunRequest struct { + RunID string + Model string + CWD string + Title string + Trigger *protocol.TriggerInfo + Prompt string + Workflow unattendedWorkflowRequest + AutoWrite bool + AutoDirs bool + JobID string + SuppressFinishBroadcast bool +} + +type unattendedRunResult struct { + SessionID string + // session is returned so callers can persist post-run metadata before their + // final session-list notification. + session *Session + FinalText string + AgentTurns int + ConfirmRequests []string + HadError bool + Err string + ErrSource string + TimedOut bool +} + +type unattendedEventPolicy func(context.Context, *Session, protocol.SessionEvent) (handled bool, err error) + +func decodeUnattendedEvent[T any](data any) T { + var out T + if data == nil { + return out + } + if typed, ok := data.(T); ok { + return typed + } + + var raw []byte + switch v := data.(type) { + case json.RawMessage: + raw = v + case []byte: + raw = v + case string: + raw = []byte(v) + default: + var err error + raw, err = json.Marshal(v) + if err != nil { + return out + } + } + _ = json.Unmarshal(raw, &out) + return out +} + +func (s *Server) runUnattendedSession(ctx context.Context, req unattendedRunRequest, policy unattendedEventPolicy) unattendedRunResult { + runID := req.RunID + if runID == "" { + runID = generateSessionID() + } + res := unattendedRunResult{SessionID: runID} + + model := req.Model + if model == "" { + model = s.model + } + session := NewSession(runID, s, nil, model, req.CWD, "", false, req.AutoWrite, req.AutoDirs, true, ctx) + res.session = session + session.origin = "vix" + session.trigger = req.Trigger + session.title = req.Title + if req.JobID != "" { + if jobsRoot := config.NewVixPaths("", s.homeVixDir, "").Jobs(); jobsRoot != "" { + session.jobDir = filepath.Join(jobsRoot, req.JobID) + session.addAllowedDir(session.jobDir) + } + } + + s.sessionMu.Lock() + s.sessions[runID] = session + s.sessionMu.Unlock() + s.broadcastSessionsChanged() + defer func() { + s.sessionMu.Lock() + delete(s.sessions, runID) + s.sessionMu.Unlock() + session.cancel() + if !req.SuppressFinishBroadcast { + s.broadcastSessionsChanged() + } + }() + + go session.Run() + + startCmd, err := unattendedStartCommand(req) + if err != nil { + res.HadError = true + res.Err = err.Error() + res.ErrSource = "start_command" + return res + } + if !session.pushCommand(ctx, startCmd) { + res.HadError = true + res.Err = "session refused start command" + res.ErrSource = "start_refused" + return res + } + + var final strings.Builder + for { + select { + case ev := <-session.eventChan: + switch ev.Type { + case "event.stream_chunk": + final.WriteString(decodeUnattendedEvent[protocol.EventStreamChunk](ev.Data).Text) + res.FinalText = final.String() + case "event.stream_done": + res.AgentTurns++ + case "event.confirm_request": + cr := decodeUnattendedEvent[protocol.EventConfirmRequest](ev.Data) + res.ConfirmRequests = append(res.ConfirmRequests, cr.ToolName) + if err := runUnattendedEventPolicy(ctx, session, ev, policy); err != nil { + res = unattendedPolicyErrorResult(ctx, res, final.String(), err) + session.persist() + return res + } + case "event.user_question", "event.plan_proposed": + if err := runUnattendedEventPolicy(ctx, session, ev, policy); err != nil { + res = unattendedPolicyErrorResult(ctx, res, final.String(), err) + session.persist() + return res + } + case "event.error": + res.HadError = true + res.Err = decodeUnattendedEvent[protocol.EventError](ev.Data).Message + res.ErrSource = "agent" + case "event.agent_done": + res.FinalText = final.String() + session.persist() + return res + } + case <-ctx.Done(): + res = cancelledUnattendedResult(res, final.String(), ctx.Err()) + session.persist() + return res + case <-session.ctx.Done(): + res.FinalText = final.String() + if err := ctx.Err(); err != nil { + res = cancelledUnattendedResult(res, final.String(), err) + session.persist() + return res + } + res.HadError = true + res.ErrSource = "session" + res.Err = "session ended before agent completion" + session.persist() + return res + } + } +} + +func cancelledUnattendedResult(res unattendedRunResult, final string, err error) unattendedRunResult { + res.FinalText = final + res.TimedOut = true + res.HadError = true + res.ErrSource = "timeout" + res.Err = "run cancelled: " + err.Error() + return res +} + +func unattendedPolicyErrorResult(ctx context.Context, res unattendedRunResult, final string, err error) unattendedRunResult { + if ctxErr := ctx.Err(); ctxErr != nil { + return cancelledUnattendedResult(res, final, ctxErr) + } + res.HadError = true + res.Err = err.Error() + res.ErrSource = "policy" + res.FinalText = final + return res +} + +func runUnattendedEventPolicy(ctx context.Context, session *Session, ev protocol.SessionEvent, policy unattendedEventPolicy) error { + if policy == nil { + return fmt.Errorf("unhandled unattended event: %s", ev.Type) + } + handled, err := policy(ctx, session, ev) + if err != nil { + return err + } + if !handled { + return fmt.Errorf("unhandled unattended event: %s", ev.Type) + } + return nil +} + +func unattendedStartCommand(req unattendedRunRequest) (protocol.SessionCommand, error) { + switch { + case req.Workflow.Inline != nil: + raw, err := json.Marshal(req.Workflow.Inline) + if err != nil { + return protocol.SessionCommand{}, err + } + data, err := json.Marshal(protocol.SessionWorkflowData{Name: req.Workflow.Inline.Name, Text: req.Prompt, Workflow: raw}) + if err != nil { + return protocol.SessionCommand{}, err + } + return protocol.SessionCommand{Type: "session.workflow", Data: data}, nil + case req.Workflow.Name != "": + data, err := json.Marshal(protocol.SessionWorkflowData{Name: req.Workflow.Name, Text: req.Prompt}) + if err != nil { + return protocol.SessionCommand{}, err + } + return protocol.SessionCommand{Type: "session.workflow", Data: data}, nil + default: + data, err := json.Marshal(protocol.SessionInputData{Text: req.Prompt}) + if err != nil { + return protocol.SessionCommand{}, err + } + return protocol.SessionCommand{Type: "session.input", Data: data}, nil + } +} diff --git a/internal/protocol/types.go b/internal/protocol/types.go index 4b1f097..e38c866 100644 --- a/internal/protocol/types.go +++ b/internal/protocol/types.go @@ -70,8 +70,8 @@ type SessionStartData struct { AttachSessionID string `json:"attach_session_id,omitempty"` } -// TriggerInfo records what fired a vix-initiated session: a scheduled job's -// trigger type ("cron" | "at") and the job id. +// TriggerInfo records what fired a vix-initiated session: a scheduled or manual +// job, a lifecycle hook, or remote control. type TriggerInfo struct { Type string `json:"type"` Ref string `json:"ref,omitempty"`