diff --git a/README.md b/README.md index 091b544f9a..0ca1655703 100644 --- a/README.md +++ b/README.md @@ -2007,6 +2007,18 @@ KCAP_CODEX_TRANSPORT=app-server kcap daemon # opt in; default is `pty` - **`app-server`** takes effect only for review-flow (unattended) launches and only when the installed Codex meets the minimum version (**0.146.0**); a lower build falls back to `pty` automatically. Interactive launches always use `pty` in this release. - Rollback is a restart with the value flipped back to `pty`; it governs new launches only. +##### Claude transport (`KCAP_CLAUDE_TRANSPORT`) + +Interactive hosted Claude agents can run as `claude -p` over Claude Code's stream-json protocol instead of the interactive PTY. The prompt is then sent as the session's first message, never on the command line, so the chat shows what you typed whatever its length. It is opt-in and env-only: + +```bash +KCAP_CLAUDE_TRANSPORT=stream-json kcap daemon # opt in; default is `pty` +``` + +- **Default `pty`** — the interactive-terminal path, unchanged. +- **`stream-json`** applies to interactive launches in a daemon-owned worktree. Reviews, review flows and private agents stay on `pty`. A stream-json agent has no terminal view; permission prompts and questions arrive as cards as usual. +- Rollback is a restart with the value flipped back to `pty`; it governs new launches only. + The Cursor CLI path (`cursor-agent` by default, used to spawn the `cursor` hosted-agent vendor) is env-only for now — there is no `daemon.cursor_path` profile key yet, so set it per-launch: ```bash diff --git a/src/Capacitor.App/App.axaml.cs b/src/Capacitor.App/App.axaml.cs index 6c1a9a4559..6f853ade62 100644 --- a/src/Capacitor.App/App.axaml.cs +++ b/src/Capacitor.App/App.axaml.cs @@ -1584,7 +1584,7 @@ internal static MainWindow BuildAndShowMainWindow( service, appState, launch, new RepoPathStore(config, time).GetSortedPathsAsync, time, shutdownToken, navigationGeneration: () => vm?.NavigationGeneration ?? 0, - openSessionIfCurrent: (agentId, generation) => vm?.OpenSessionIfCurrent(agentId, generation), + openSessionIfCurrent: (agentId, generation, launchPrompt) => vm?.OpenSessionIfCurrent(agentId, generation, launchPrompt), requestSignIn: requestSignIn, daemons: remoteAgents?.Daemons, viewerId: viewerId, laneStatus: lane?.Status, localMachineId: localMachineId, launchFailures: lane?.LaunchFailures, directory: resolvedDirectory, diff --git a/src/Capacitor.App/ViewModels/ChatTabViewModel.cs b/src/Capacitor.App/ViewModels/ChatTabViewModel.cs index 4215e77778..c36f55bab7 100644 --- a/src/Capacitor.App/ViewModels/ChatTabViewModel.cs +++ b/src/Capacitor.App/ViewModels/ChatTabViewModel.cs @@ -51,6 +51,9 @@ public sealed class ChatTabViewModel : ReactiveObject, IAttachmentSink { // token it can ask without an ObjectDisposedException. readonly CancellationToken _lifetimeToken; readonly AvaloniaList _items = new(); + /// Stands in for the launch prompt until the transcript's own rows land. Never a queued send: + /// the transcript's first read is history, which cannot acknowledge one. + UserTurnItem? _launchTurn; readonly AvaloniaList _queuedMessages = new(); // Keep retired sends eligible for late echoes without letting them consume a later send's receipt. readonly List<(QueuedChatMessage Old, QueuedChatMessage Replacement)> _superseded = new(); @@ -1020,10 +1023,18 @@ async Task ChooseTerminalMenuAsync(byte[] keys) { SetTerminalMenuChoiceTaken(false); } + public void ShowLaunchPrompt(string text) { + if (_launchTurn is not null || _items.Count > 0) return; + _launchTurn = new UserTurnItem(text); + _items.Add(_launchTurn); + this.RaisePropertyChanged(nameof(PhaseNote)); + } + void SwitchFeed(string key, Func open) { // A released tab's runs belong to the monitor that took its feed; this tab reads nothing more. if (_feedReleased) return; _items.Clear(); + if (_launchTurn is not null) _items.Add(_launchTurn); _history.Clear(); _literalInputHistory.Clear(); _unechoed.Clear(); @@ -1339,6 +1350,7 @@ bool Prepare(int generation, FeedRead read) { // names where that history ends — anything past it arrived after and can acknowledge. RebaseQueuedMessages(read.SnapshotOffset ?? CurrentOffset ?? 0); _items.Clear(); + if (_launchTurn is not null) _items.Add(_launchTurn); _history.ClearForReplay(); _pendingTools.Clear(); _settledTools.Clear(); @@ -1485,7 +1497,11 @@ void Finish(FeedRead read, LineCursor? cursor) { foreach (var projected in cursor.Restored) foreach (var text in projected.SubmittedInputs) _history.Observe(AttachmentTrailer.Strip(text)); - if (cursor.Fresh.Count > 0) _items.AddRange(cursor.Fresh); + if (cursor.Fresh.Count > 0) { + if (_launchTurn is not null) _items.Remove(_launchTurn); + _launchTurn = null; + _items.AddRange(cursor.Fresh); + } } Phase = ChatTabPhase.Reading; ApplyFeedRuns(read); diff --git a/src/Capacitor.App/ViewModels/HomeViewModel.cs b/src/Capacitor.App/ViewModels/HomeViewModel.cs index edcbd27843..27d487349a 100644 --- a/src/Capacitor.App/ViewModels/HomeViewModel.cs +++ b/src/Capacitor.App/ViewModels/HomeViewModel.cs @@ -114,7 +114,7 @@ internal static string NoHarnessesNotice(string? remoteMachine) => remoteMachine readonly ILaunchClient _launch; readonly Func> _knownRepos; readonly Func? _navigationGeneration; - readonly Action? _openSessionIfCurrent; + readonly Action? _openSessionIfCurrent; readonly Action? _launchFailed; readonly CompositeDisposable _disposables = new(); @@ -470,7 +470,7 @@ public HomeViewModel( IDaemonClientService daemon, IAppStateStore state, ILaunchClient launch, Func> knownRepos, TimeProvider time, CancellationToken shutdown = default, Func? navigationGeneration = null, - Action? openSessionIfCurrent = null, Action? requestSignIn = null, + Action? openSessionIfCurrent = null, Action? requestSignIn = null, IObservable>? daemons = null, Func>? viewerId = null, IObservable? laneStatus = null, string? localMachineId = null, @@ -1372,7 +1372,7 @@ async Task StartAsync() { // Untracked with no row is a failure that beat the call's own return: opening would cover // the launcher that has just worded it. if (!draft.Remote && (tracked || RowExists(agentId, AgentOrigin.Local))) - _openSessionIfCurrent?.Invoke(agentId, generation); + _openSessionIfCurrent?.Invoke(agentId, generation, draft.Goal); } async Task OwnsConnectedAsync(string machine) => diff --git a/src/Capacitor.App/ViewModels/MainWindowViewModel.cs b/src/Capacitor.App/ViewModels/MainWindowViewModel.cs index 36eae092d5..6adc14f6fc 100644 --- a/src/Capacitor.App/ViewModels/MainWindowViewModel.cs +++ b/src/Capacitor.App/ViewModels/MainWindowViewModel.cs @@ -672,9 +672,10 @@ open is RemoteSessionViewModel remote /// arriving after any navigation (closing the workspace, another session, close-to-hide, the /// shutdown latch) opens nothing, rather than attaching an invisible terminal or replacing what /// the user opened while the launch was in flight. - public void OpenSessionIfCurrent(string agentId, int generation) { + public void OpenSessionIfCurrent(string agentId, int generation, string? launchPrompt = null) { if (generation != _navigation.Generation) return; - OpenSession(agentId); + if (OpenSession(agentId) && launchPrompt is { Length: > 0 } && CurrentWorkspace is WorkspaceViewModel workspace) + workspace.ShowLaunchPrompt(launchPrompt); } /// The launcher words a launch failure, and an open workspace covers the launcher. Only a diff --git a/src/Capacitor.App/ViewModels/WorkspaceViewModel.cs b/src/Capacitor.App/ViewModels/WorkspaceViewModel.cs index d826a2c0f7..370d3c052e 100644 --- a/src/Capacitor.App/ViewModels/WorkspaceViewModel.cs +++ b/src/Capacitor.App/ViewModels/WorkspaceViewModel.cs @@ -56,6 +56,13 @@ public sealed class WorkspaceViewModel : ReactiveObject, ISessionWorkspace { public IObservable Ended { get; } + string? _launchPrompt; + + public void ShowLaunchPrompt(string prompt) { + if (Chat is { } chat) chat.ShowLaunchPrompt(prompt); + else _launchPrompt = prompt; + } + ChatTabViewModel? _chat; /// Built once, on the first dto; the reader comes from transcript_format and the input channel /// from has_terminal. @@ -331,6 +338,7 @@ public WorkspaceViewModel( }; if (PullRequests is { } fixHost) Chat.WhenAnyValue(chat => chat.ShowsComposer).Subscribe(_ => fixHost.FixAvailabilityChanged()).DisposeWith(_disposables); + if (_launchPrompt is { } launchPrompt) Chat.ShowLaunchPrompt(launchPrompt); }) .DisposeWith(_disposables); diff --git a/src/Capacitor.Cli.Daemon/DaemonConfig.cs b/src/Capacitor.Cli.Daemon/DaemonConfig.cs index d19ffa4d1c..17b5e6d701 100644 --- a/src/Capacitor.Cli.Daemon/DaemonConfig.cs +++ b/src/Capacitor.Cli.Daemon/DaemonConfig.cs @@ -216,6 +216,10 @@ public BinaryProbe Binaries { public string ClaudePath { get; set; } = Core.Harness.Claude.ClaudeHarness.CliBinary; public string CodexPath { get; set; } = Core.Harness.Codex.CodexHarness.CliBinary; + /// Transport for interactive hosted Claude agents, from KCAP_CLAUDE_TRANSPORT: + /// pty (default) or stream-json. Any other value is pty. + public string ClaudeTransport { get; set; } = Harness.Claude.ClaudeHostedAgentRuntimeFactory.Pty; + /// Transport for hosted Codex reviewers: pty (default) or app-server. /// Set via KCAP_CODEX_TRANSPORT. Governs NEW review-flow launches only; interactive /// launches always take the PTY path in this phase. The effective decision (this selection AND diff --git a/src/Capacitor.Cli.Daemon/DaemonRunner.cs b/src/Capacitor.Cli.Daemon/DaemonRunner.cs index abf92187f1..eda8de0fbf 100644 --- a/src/Capacitor.Cli.Daemon/DaemonRunner.cs +++ b/src/Capacitor.Cli.Daemon/DaemonRunner.cs @@ -209,6 +209,9 @@ public static async Task RunAsync(string[] args) { Harness.Codex.CodexTransportDecision.BindFromEnvironment(config, Environment.GetEnvironmentVariable); + if (Environment.GetEnvironmentVariable("KCAP_CLAUDE_TRANSPORT") is { Length: > 0 } claudeTransport) + config.ClaudeTransport = claudeTransport.Trim(); + // One resolved fact for both the launch router and the certification advertisement. config.CodexAppServerActive = Harness.Codex.CodexTransportDecision.ResolveActive( config.CodexTransport, () => ProbeCliVersion(config.CodexPath)); @@ -457,15 +460,22 @@ await Console.Error.WriteLineAsync( // Prepare/BuildArgs/Spawn inline. PtyHostedAgentRuntimeFactory wraps each registered PTY // launcher (Claude, Codex); AcpHostedAgentRuntimeFactory speaks ACP JSON-RPC over stdio for // Cursor (no IHostedAgentLauncher — Cursor never went through the PTY launcher contract). - builder.Services.AddSingleton(sp => - new PtyHostedAgentRuntimeFactory( - sp.GetServices().SingleOrDefault(l => l.Vendor == "claude") - ?? throw new InvalidOperationException("No IHostedAgentLauncher registered for vendor 'claude'"), + builder.Services.AddSingleton(sp => { + var claudeLauncher = sp.GetServices().SingleOrDefault(l => l.Vendor == "claude") + ?? throw new InvalidOperationException("No IHostedAgentLauncher registered for vendor 'claude'"); + var pty = new PtyHostedAgentRuntimeFactory( + claudeLauncher, sp.GetRequiredService(), sp.GetRequiredService>(), - sp.GetRequiredService() - ) - ); + sp.GetRequiredService()); + return new Harness.Claude.ClaudeHostedAgentRuntimeFactory( + claudeLauncher, pty, + sp.GetRequiredService(), + sp.GetRequiredService(), + harnesses.Of().Paths, + sp.GetRequiredService(), + time); + }); // Codex is the ONE vendor with two transports. The composite factory routes review-flow // launches to codex app-server when this daemon resolved it active (config.CodexAppServerActive), // and delegates everything else — every interactive launch, and all launches under the PTY diff --git a/src/Capacitor.Cli.Daemon/Harness/Claude/ClaudeHostedAgentRuntimeFactory.cs b/src/Capacitor.Cli.Daemon/Harness/Claude/ClaudeHostedAgentRuntimeFactory.cs new file mode 100644 index 0000000000..9b58494efc --- /dev/null +++ b/src/Capacitor.Cli.Daemon/Harness/Claude/ClaudeHostedAgentRuntimeFactory.cs @@ -0,0 +1,162 @@ +using System.Diagnostics; +using System.Text; +using Capacitor.Cli.Core; +using Capacitor.Cli.Core.Harness.Claude; +using Capacitor.Cli.Core.LocalIpc; +using Capacitor.Cli.Daemon.Pty; +using Capacitor.Cli.Daemon.Services; +using Microsoft.Extensions.Logging; + +namespace Capacitor.Cli.Daemon.Harness.Claude; + +/// +/// Hosts Claude on one of two transports. Under stream-json an interactive launch runs +/// claude -p and gets its prompt as its first stdin message, so the prompt never reaches the +/// command line; every other launch, and every launch under pty, goes to the PTY factory. +/// +internal sealed partial class ClaudeHostedAgentRuntimeFactory : IHostedAgentRuntimeFactory { + public const string Pty = "pty"; + public const string StreamJson = "stream-json"; + + static readonly string[] StreamJsonArgs = [ + "-p", "--input-format", "stream-json", "--output-format", "stream-json", "--verbose", + "--permission-prompt-tool", "stdio" + ]; + + readonly IHostedAgentLauncher _launcher; + readonly IHostedAgentRuntimeFactory _pty; + readonly DaemonConfig _config; + readonly LocalPermissionBridge _bridge; + readonly ClaudePaths _paths; + readonly ILoggerFactory _loggerFactory; + readonly TimeProvider _time; + readonly ILogger _logger; + readonly Func _processSource; + + /// Replaces the real spawn; production passes null. + public ClaudeHostedAgentRuntimeFactory( + IHostedAgentLauncher launcher, + IHostedAgentRuntimeFactory ptyDelegate, + DaemonConfig config, + LocalPermissionBridge bridge, + ClaudePaths paths, + ILoggerFactory loggerFactory, + TimeProvider time, + Func? processSource = null) { + _launcher = launcher; + _pty = ptyDelegate; + _config = config; + _bridge = bridge; + _paths = paths; + _loggerFactory = loggerFactory; + _time = time; + _logger = loggerFactory.CreateLogger(); + _processSource = processSource ?? (psi => new JsonLineProcess( + WindowsLaunchCommand.Apply(psi, config.Binaries), loggerFactory.CreateLogger(), time)); + } + + public string Vendor => "claude"; + + public string CliPath => _pty.CliPath; + public bool IsAvailable() => _pty.IsAvailable(); + public bool SupportsUnattended => _pty.SupportsUnattended; + public UnattendedSupport DescribeUnattendedSupport() => _pty.DescribeUnattendedSupport(); + public bool SupportsPrReview => _pty.SupportsPrReview; + public bool SupportsBorrowedReviewFlow => _pty.SupportsBorrowedReviewFlow; + public string? BorrowedReviewContainment => _pty.BorrowedReviewContainment; + public bool BorrowedReviewRequiresIndependentSnapshot => _pty.BorrowedReviewRequiresIndependentSnapshot; + public bool ReviewFlowRedirectsHome => _pty.ReviewFlowRedirectsHome; + public IReviewerModelResolver? ReviewerModelResolver => _pty.ReviewerModelResolver; + public bool SupportsModelSelection => _pty.SupportsModelSelection; + public bool SupportsRestartResume(string? runtimeTransport) => _pty.SupportsRestartResume(runtimeTransport); + + public Task?> ProbeModelsAsync(CancellationToken ct) => _pty.ProbeModelsAsync(ct); + public IReadOnlyList CatalogFingerprintPaths => _pty.CatalogFingerprintPaths; + public AttachmentPlacement AttachmentPlacementFor(LaunchKind kind) => _pty.AttachmentPlacementFor(kind); + + /// Reviews keep the PTY, and so does a private agent, which has no bridge to answer its + /// prompts. Without a bridge URL the hook would raise its own card beside the runtime's. + internal bool UsesStreamJson(RuntimeStartContext ctx) => + _config.ClaudeTransport == StreamJson + && !ctx.IsReview && !ctx.IsReviewFlow && !ctx.IsPrivate + && ctx.Work == WorkLocation.OwnedWorktree + && !string.IsNullOrEmpty(ctx.DaemonBridgeUrl); + + public Task StartAsync(RuntimeStartContext ctx, CancellationToken ct) => + UsesStreamJson(ctx) ? StartStreamJsonAsync(ctx, ct) : _pty.StartAsync(ctx, ct); + + async Task StartStreamJsonAsync(RuntimeStartContext ctx, CancellationToken ct) { + var launcherCtx = await PtyHostedAgentRuntimeFactory.PrepareAsync(_launcher, ctx, ex => LogPrepareSoftFailure(ex, ctx.AgentId)); + var launchArgs = _launcher.BuildArgs(launcherCtx with { Prompt = null }); + + LogLaunching(ctx.AgentId, ctx.Worktree.Path); + + var process = _processSource(BuildPsi(_launcher.CliPath, [.. launchArgs.Args, .. StreamJsonArgs], ctx)); + var runtime = new ClaudeStreamJsonHostedAgentRuntime( + process, + (request, token) => _bridge.DecideAsync(request, Vendor, token), + sessionId => Path.Combine(_paths.ProjectDir(ctx.Worktree.Path), sessionId + ".jsonl"), + ctx.AgentId, + ctx.Worktree.Path, + _loggerFactory.CreateLogger(), + _time); + + try { + ct.ThrowIfCancellationRequested(); + await runtime.SendInitializeAsync().ConfigureAwait(false); + if (!string.IsNullOrEmpty(ctx.Prompt)) await runtime.SendUserInputAsync(ctx.Prompt).ConfigureAwait(false); + } catch (Exception ex) { + var stderr = process.Diagnostics; + await runtime.DisposeAsync().ConfigureAwait(false); + + if (ex is OperationCanceledException && ct.IsCancellationRequested) throw; + + // stderr goes to the daemon log only: this exception is forwarded to the server. + if (stderr is { Length: > 0 }) { + LogLaunchStderr(stderr); + throw new InvalidOperationException($"{ex.Message} (stderr captured in daemon log)", ex); + } + + throw; + } + + return new HostedRuntimeStart(runtime, launchArgs.McpConfigPath); + } + + /// The child gets what a PTY child gets: the daemon's environment without its session, + /// supervision and boot-local variables, plus the hosted agent's own. + internal static ProcessStartInfo BuildPsi(string cliPath, IReadOnlyList args, RuntimeStartContext ctx) { + var noBomUtf8 = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false); + + var psi = new ProcessStartInfo(cliPath) { + WorkingDirectory = ctx.Worktree.Path, + RedirectStandardInput = true, + RedirectStandardOutput = true, + RedirectStandardError = true, + StandardInputEncoding = noBomUtf8, + StandardOutputEncoding = noBomUtf8, + StandardErrorEncoding = noBomUtf8, + UseShellExecute = false, + CreateNoWindow = true, + }; + + foreach (var arg in args) psi.ArgumentList.Add(arg); + + foreach (var key in PtyEnvScrub.ClaudeSessionVars) psi.Environment.Remove(key); + foreach (var key in PtyEnvScrub.HostedAgentVars) psi.Environment.Remove(key); + foreach (var key in PtyEnvScrub.DaemonSupervisionVars) psi.Environment.Remove(key); + foreach (var key in DaemonRunner.BootCarriers.All) psi.Environment.Remove(key); + foreach (var (key, value) in PtyHostedAgentRuntimeFactory.HostedEnv(ctx)) psi.Environment[key] = value; + + return psi; + } + + [LoggerMessage(Level = LogLevel.Information, Message = "Claude stream-json launch: agentId={AgentId} cwd={Cwd}")] + partial void LogLaunching(string agentId, string cwd); + + [LoggerMessage(Level = LogLevel.Warning, Message = "Launcher Prepare soft-failure for agent {AgentId} (continuing)")] + partial void LogPrepareSoftFailure(Exception ex, string agentId); + + [LoggerMessage(Level = LogLevel.Warning, Message = "Claude stream-json launch failed; captured stderr follows (not sent to server): {Diagnostics}")] + partial void LogLaunchStderr(string diagnostics); +} diff --git a/src/Capacitor.Cli.Daemon/Harness/Claude/ClaudeStreamJsonHostedAgentRuntime.cs b/src/Capacitor.Cli.Daemon/Harness/Claude/ClaudeStreamJsonHostedAgentRuntime.cs new file mode 100644 index 0000000000..473f9da1f5 --- /dev/null +++ b/src/Capacitor.Cli.Daemon/Harness/Claude/ClaudeStreamJsonHostedAgentRuntime.cs @@ -0,0 +1,297 @@ +using System.Collections.Concurrent; +using System.Runtime.CompilerServices; +using System.Text.Json; +using System.Text.Json.Nodes; +using Capacitor.Cli.Core; +using Capacitor.Cli.Daemon.Services; +using Microsoft.Extensions.Logging; + +namespace Capacitor.Cli.Daemon.Harness.Claude; + +/// +/// Claude Code driven over its stream-json protocol: every user message, the first included, is a +/// JSON line on stdin, and a permission prompt or question arrives as a can_use_tool control +/// request answered on stdin. The hooks and the transcript watcher still record the session, so +/// nothing on stdout is forwarded. +/// +internal sealed partial class ClaudeStreamJsonHostedAgentRuntime : IHostedAgentRuntime { + const string DeniedMessage = "The user denied this tool use."; + + readonly IJsonLineProcess _process; + readonly Func> _decide; + readonly Func _transcriptPathOf; + readonly string _agentId; + readonly string _cwd; + readonly ILogger _logger; + readonly TimeProvider _time; + readonly CancellationTokenSource _lifetime = new(); + readonly ConcurrentDictionary _asks = new(); + readonly Task _readLoop; + + volatile string? _sessionId; + int _inputClosed; + int _disposed; + + /// Answers one permission request, given in the shape the hook posts to the + /// bridge. + /// The transcript a session id is recorded in, for the judge. + internal ClaudeStreamJsonHostedAgentRuntime( + IJsonLineProcess process, + Func> decide, + Func transcriptPathOf, + string agentId, + string cwd, + ILogger logger, + TimeProvider time) { + _process = process; + _decide = decide; + _transcriptPathOf = transcriptPathOf; + _agentId = agentId; + _cwd = cwd; + _logger = logger; + _time = time; + _readLoop = ReadLoopAsync(); + } + + public string Vendor => "claude"; + public int Pid => _process.Pid; + public bool HasExited => _process.HasExited; + public int? ExitCode => _process.ExitCode; + public bool EmitsTerminalOutput => false; + public bool AsksPermissionsOnStdin => true; + + /// Null until the file exists: a reader linked to a path not yet written reports the transcript missing. + internal (string SessionId, string Path)? KnownTranscript => + _sessionId is { } id && _transcriptPathOf(id) is { } path && File.Exists(path) ? (id, path) : null; + + /// Holds the orchestrator's read loop open until the child's stdout ends. + public async IAsyncEnumerable ReadOutputAsync([EnumeratorCancellation] CancellationToken ct = default) { + await _readLoop.WaitAsync(ct).ConfigureAwait(false); + yield break; + } + + public Task SendUserInputAsync(string text) { + if (Volatile.Read(ref _inputClosed) != 0) throw new InputNotAdmittedException("Claude: the hosted session has ended."); + + return WriteAsync(UserMessage(text)); + } + + /// What the Agent SDK sends before its first message; the launch is verified this way. + internal Task SendInitializeAsync() => WriteAsync(ControlRequest("initialize")); + + public async Task SendSpecialKeyAsync(string key) { + if (!string.Equals(key, "escape", StringComparison.OrdinalIgnoreCase)) return; + + try { + await WriteAsync(ControlRequest("interrupt")).ConfigureAwait(false); + } catch (Exception ex) { + LogInterruptFailed(ex, _agentId); + } + } + + public Task SendRawInputAsync(byte[] data) => + throw new NotSupportedException("Local-attach raw input is a PTY-only surface; this Claude runtime has no terminal."); + + public void Resize(ushort cols, ushort rows) { } + + /// Interrupts the running turn and closes stdin, which ends the session once the turn + /// stops. + public async Task RequestGracefulStopAsync() { + await SendSpecialKeyAsync("escape").ConfigureAwait(false); + Volatile.Write(ref _inputClosed, 1); + await _process.CloseInputAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); + } + + public Task WaitForExitAsync(TimeSpan? timeout = null) => _process.WaitForExitAsync(timeout); + + public async Task TerminateAsync(TimeSpan? timeout = null) { + Volatile.Write(ref _inputClosed, 1); + await _lifetime.CancelAsync().ConfigureAwait(false); + await _process.TerminateAsync(timeout).ConfigureAwait(false); + } + + public async ValueTask DisposeAsync() { + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; + + Volatile.Write(ref _inputClosed, 1); + await _lifetime.CancelAsync().ConfigureAwait(false); + await _process.DisposeAsync().ConfigureAwait(false); + + try { + await _readLoop.WaitAsync(TimeSpan.FromSeconds(2), _time).ConfigureAwait(false); + } catch { + } + + _lifetime.Dispose(); + } + + async Task ReadLoopAsync() { + await Task.Yield(); + + try { + await foreach (var line in _process.ReadLinesAsync(_lifetime.Token).ConfigureAwait(false)) { + JsonElement message; + + try { + using var doc = JsonDocument.Parse(line); + message = doc.RootElement.Clone(); + } catch (JsonException) { + continue; + } + + if (message.Str("session_id") is { Length: > 0 } sessionId) _sessionId = sessionId; + + switch (message.Str("type")) { + case "control_request": + OnControlRequest(message); + break; + case "control_cancel_request": + if (message.Str("request_id") is { } cancelled && _asks.TryRemove(cancelled, out var ask)) CancelQuietly(ask); + break; + } + } + } catch (OperationCanceledException) { + } catch (Exception ex) { + LogReadLoopFailed(ex, _agentId); + } finally { + Volatile.Write(ref _inputClosed, 1); + foreach (var ask in _asks.Values) CancelQuietly(ask); + } + } + + void OnControlRequest(JsonElement message) { + if (message.Str("request_id") is not { } requestId || message.Obj("request") is not { } request) return; + + if (request.Str("subtype") == "can_use_tool") { + _ = AnswerAsync(requestId, request); + return; + } + + _ = WriteQuietlyAsync(new JsonObject { + ["type"] = "control_response", + ["response"] = new JsonObject { + ["subtype"] = "error", + ["request_id"] = requestId, + ["error"] = $"unsupported control request: {request.Str("subtype")}" + } + }); + } + + async Task AnswerAsync(string requestId, JsonElement request) { + using var ask = CancellationTokenSource.CreateLinkedTokenSource(_lifetime.Token); + _asks[requestId] = ask; + + try { + PermissionDecision decision; + + if (_sessionId is not { } sessionId) { + decision = PermissionSettlements.DenyDecision; + } else { + try { + decision = await _decide(BridgeRequest(sessionId, request), ask.Token).ConfigureAwait(false); + } catch (Exception ex) when (ex is not OperationCanceledException) { + LogDecisionFailed(ex, _agentId); + decision = PermissionSettlements.DenyDecision; + } + } + + if (ask.IsCancellationRequested) return; + + await WriteAsync(PermissionResponse(requestId, decision, request.Prop("input"))).ConfigureAwait(false); + } catch (OperationCanceledException) { + } catch (Exception ex) { + LogAnswerFailed(ex, _agentId); + } finally { + _asks.TryRemove(new KeyValuePair(requestId, ask)); + } + } + + /// The request in the shape Claude's permission hook posts to the bridge, so one decision + /// path serves both. + JsonObject BridgeRequest(string sessionId, JsonElement request) { + var payload = new JsonObject { + ["session_id"] = sessionId, + ["tool_name"] = request.Str("tool_name") ?? "Unknown", + ["tool_input"] = NodeOf(request.Prop("input")), + ["permission_suggestions"] = NodeOf(request.Prop("permission_suggestions")), + ["agent_id"] = _agentId, + ["cwd"] = _cwd + }; + + if (request.Str("tool_use_id") is { } toolUseId) payload["tool_use_id"] = toolUseId; + if (_transcriptPathOf(sessionId) is { } transcript) payload["transcript_path"] = transcript; + if (request.Str("agent_id") is { Length: > 0 } subagentId) payload["subagent_id"] = subagentId; + + return payload; + } + + internal static JsonObject PermissionResponse(string requestId, PermissionDecision decision, JsonElement? input) { + var result = decision.Behavior == PermissionSettlements.Allow + ? new JsonObject { + ["behavior"] = "allow", + ["updatedInput"] = NodeOf(decision.UpdatedInput ?? input) ?? new JsonObject() + } + : new JsonObject { + ["behavior"] = "deny", + ["message"] = DeniedMessage + }; + + if (decision.Behavior == PermissionSettlements.Allow && decision.ApplyPermissions is { ValueKind: JsonValueKind.Array } rules) + result["updatedPermissions"] = NodeOf(rules); + + return new JsonObject { + ["type"] = "control_response", + ["response"] = new JsonObject { + ["subtype"] = "success", + ["request_id"] = requestId, + ["response"] = result + } + }; + } + + internal static JsonObject UserMessage(string text) => new() { + ["type"] = "user", + ["message"] = new JsonObject { ["role"] = "user", ["content"] = text }, + ["parent_tool_use_id"] = null, + ["session_id"] = "" + }; + + internal static JsonObject ControlRequest(string subtype) => new() { + ["type"] = "control_request", + ["request_id"] = Guid.NewGuid().ToString("N"), + ["request"] = new JsonObject { ["subtype"] = subtype } + }; + + /// The answer that owns may finish and dispose it first. + static void CancelQuietly(CancellationTokenSource ask) { + try { + ask.Cancel(); + } catch (ObjectDisposedException) { + } + } + + static JsonNode? NodeOf(JsonElement? element) => + element is { } e && !e.IsNull ? JsonNode.Parse(e.GetRawText()) : null; + + Task WriteAsync(JsonNode message) => _process.WriteLineAsync(message.ToJsonString(), _lifetime.Token); + + async Task WriteQuietlyAsync(JsonNode message) { + try { + await WriteAsync(message).ConfigureAwait(false); + } catch (Exception ex) { + LogAnswerFailed(ex, _agentId); + } + } + + [LoggerMessage(Level = LogLevel.Warning, Message = "Claude stream-json: reading the agent's output failed (agentId={AgentId})")] + partial void LogReadLoopFailed(Exception ex, string agentId); + + [LoggerMessage(Level = LogLevel.Warning, Message = "Claude stream-json: deciding a permission request failed; denying it (agentId={AgentId})")] + partial void LogDecisionFailed(Exception ex, string agentId); + + [LoggerMessage(Level = LogLevel.Debug, Message = "Claude stream-json: answering a control request failed (agentId={AgentId})")] + partial void LogAnswerFailed(Exception ex, string agentId); + + [LoggerMessage(Level = LogLevel.Debug, Message = "Claude stream-json: the interrupt was not written (agentId={AgentId})")] + partial void LogInterruptFailed(Exception ex, string agentId); +} diff --git a/src/Capacitor.Cli.Daemon/Harness/Muse/MuseHostedAgentRuntimeFactory.cs b/src/Capacitor.Cli.Daemon/Harness/Muse/MuseHostedAgentRuntimeFactory.cs index 7d2c2240eb..d1a1189d7e 100644 --- a/src/Capacitor.Cli.Daemon/Harness/Muse/MuseHostedAgentRuntimeFactory.cs +++ b/src/Capacitor.Cli.Daemon/Harness/Muse/MuseHostedAgentRuntimeFactory.cs @@ -46,7 +46,7 @@ internal sealed partial class MuseHostedAgentRuntimeFactory : IHostedAgentRuntim readonly bool _posixHost; readonly ILogger _logger; - readonly Func> _processSource; + readonly Func> _processSource; readonly Func _binaryExists; readonly Func _resolveVersion; readonly Func _hooksInstalled; @@ -66,7 +66,7 @@ public MuseHostedAgentRuntimeFactory( MusePaths paths, PermissionPromptBroker? permissionBroker = null, PermissionDecisionLog? permissionDecisionLog = null, - Func>? processSource = null, + Func>? processSource = null, Func? binaryExists = null, Func? resolveVersion = null, bool? posixHost = null, @@ -86,8 +86,8 @@ public MuseHostedAgentRuntimeFactory( _posixHost = posixHost ?? !OperatingSystem.IsWindows(); _logger = loggerFactory.CreateLogger(); - _processSource = processSource ?? ((psi, _) => Task.FromResult( - new MuseMspProcess(WindowsLaunchCommand.Apply(psi, config.Binaries), loggerFactory.CreateLogger(), time))); + _processSource = processSource ?? ((psi, _) => Task.FromResult( + new JsonLineProcess(WindowsLaunchCommand.Apply(psi, config.Binaries), loggerFactory.CreateLogger(), time))); _binaryExists = binaryExists ?? (path => config.Binaries.Finds(path)); _resolveVersion = resolveVersion ?? (path => new VendorVersionResolver(config.Binaries).Resolve(path)); _hooksInstalled = hooksInstalled ?? MuseHooksInstaller.IsInstalled; @@ -188,7 +188,7 @@ public async Task StartAsync(RuntimeStartContext ctx, Cancel var relay = !_hooksInstalled(_paths.SettingsJson); var relayedStart = false; - IMuseMspProcess? process = null; + IJsonLineProcess? process = null; MuseMspHostedAgentRuntime? runtime = null; try { diff --git a/src/Capacitor.Cli.Daemon/Harness/Muse/MuseMspHostedAgentRuntime.cs b/src/Capacitor.Cli.Daemon/Harness/Muse/MuseMspHostedAgentRuntime.cs index 508009ce70..2da6cbba5a 100644 --- a/src/Capacitor.Cli.Daemon/Harness/Muse/MuseMspHostedAgentRuntime.cs +++ b/src/Capacitor.Cli.Daemon/Harness/Muse/MuseMspHostedAgentRuntime.cs @@ -34,7 +34,7 @@ internal sealed partial class MuseMspHostedAgentRuntime : IHostedAgentRuntime, I const int ApprovalAlreadyResolvedCode = -32051; - readonly IMuseMspProcess _process; + readonly IJsonLineProcess _process; readonly ILogger _logger; readonly string _agentId; readonly string _cwd; @@ -106,7 +106,7 @@ internal sealed partial class MuseMspHostedAgentRuntime : IHostedAgentRuntime, I volatile bool _sessionClosedSeen; public MuseMspHostedAgentRuntime( - IMuseMspProcess process, + IJsonLineProcess process, ILogger logger, string agentId, string cwd, diff --git a/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs b/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs index 0dc07f6c5d..9db61a9ee4 100644 --- a/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs +++ b/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs @@ -1132,6 +1132,9 @@ internal void SetCommands(AgentInstance agent, IReadOnlyList static bool FreshIn(AgentInstance agent, string canonicalSession) => agent.BridgedSinceStart && agent.SessionId is { } own && PermissionWire.Canonical(own) == canonicalSession; + static AttributedAgent Attributed(AgentInstance agent, string canonicalSession) => + new(agent.Id, agent.PolicySnapshot, FreshIn(agent, canonicalSession), agent.Runtime.AsksPermissionsOnStdin); + /// The attribution ladder: the payload's agent id (raw, then canonical GUID), the resolved /// vendor session id, the worktree path — each rung only on exactly one live match. Live is /// "present in _agents"; teardown withdraws whatever was attributed during that window. @@ -1143,21 +1146,21 @@ static bool FreshIn(AgentInstance agent, string canonicalSession) => if (query.AgentId is { Length: > 0 } rawId) { var raw = live.Where(a => string.Equals(a.Id, rawId, StringComparison.Ordinal)).ToList(); - if (raw.Count == 1) return new AttributedAgent(raw[0].Id, raw[0].PolicySnapshot, FreshIn(raw[0], canonicalSession)); + if (raw.Count == 1) return Attributed(raw[0], canonicalSession); if (PermissionWire.Canonical(rawId) is { } canonicalId) { var canon = live.Where(a => PermissionWire.Canonical(a.Id) == canonicalId).ToList(); - if (canon.Count == 1) return new AttributedAgent(canon[0].Id, canon[0].PolicySnapshot, FreshIn(canon[0], canonicalSession)); + if (canon.Count == 1) return Attributed(canon[0], canonicalSession); } } var bySession = live.Where(a => a.SessionId is { } s && PermissionWire.Canonical(s) == canonicalSession).ToList(); - if (bySession.Count == 1) return new AttributedAgent(bySession[0].Id, bySession[0].PolicySnapshot, FreshIn(bySession[0], canonicalSession)); + if (bySession.Count == 1) return Attributed(bySession[0], canonicalSession); if (query.Cwd is { Length: > 0 } cwd) { var wanted = Path.TrimEndingDirectorySeparator(cwd); var byCwd = live.Where(a => string.Equals( Path.TrimEndingDirectorySeparator(a.Worktree.Path), wanted, RepoPathStore.PathComparison)).ToList(); - if (byCwd.Count == 1) return new AttributedAgent(byCwd[0].Id, byCwd[0].PolicySnapshot, FreshIn(byCwd[0], canonicalSession)); + if (byCwd.Count == 1) return Attributed(byCwd[0], canonicalSession); } return null; @@ -5742,6 +5745,8 @@ internal static string MapLaunchFailureReason(string? reason, string fallbackSou static readonly TimeSpan SessionIdPollInterval = TimeSpan.FromSeconds(2); + static readonly TimeSpan KnownTranscriptPollInterval = TimeSpan.FromMilliseconds(100); + // Codex turn-start diagnostic: how long, and how often, to watch a hosted Codex reviewer's rollout for growth // after a round's input is delivered, so the daemon log can say whether Codex began a turn. // A working reviewer appends response items within seconds of starting; 2 minutes of silence @@ -5756,7 +5761,9 @@ internal static string MapLaunchFailureReason(string? reason, string fallbackSou /// locator is a no-op. Best-effort background work, cancelled with the agent; it never blocks /// a launch. Task DetectSessionIdAsync(AgentInstance agent, string vendor, DateTime spawnedAtUtc) { + var streamJson = agent.Runtime as ClaudeStreamJsonHostedAgentRuntime; Func, (string SessionId, string Path)?>? locate = vendor.ToLowerInvariant() switch { + "claude" when streamJson is not null => _ => streamJson.KnownTranscript, "claude" => ruledOut => SessionTranscriptLocator.TryLocateWinner( _harnesses.Of().Paths.ProjectDir(agent.Worktree.Path), agent.Worktree.Path, spawnedAtUtc, ruledOut), "codex" => ruledOut => CodexSessionRolloutLocator.TryLocateWinner( @@ -5769,22 +5776,22 @@ Task DetectSessionIdAsync(AgentInstance agent, string vendor, DateTime spawnedAt if (locate is null) return Task.CompletedTask; Interlocked.Increment(ref _discoveryStarts); - var task = RunDiscoveryAsync(agent, locate); + var task = RunDiscoveryAsync(agent, locate, streamJson is null ? SessionIdPollInterval : KnownTranscriptPollInterval); agent.DiscoveryTask = task; return task; } internal Task RunDiscoveryForTest(AgentInstance agent, Func, (string SessionId, string Path)?> locate) { - var task = RunDiscoveryAsync(agent, locate); + var task = RunDiscoveryAsync(agent, locate, SessionIdPollInterval); agent.DiscoveryTask = task; return task; } - async Task RunDiscoveryAsync(AgentInstance agent, Func, (string SessionId, string Path)?> locate) { + async Task RunDiscoveryAsync(AgentInstance agent, Func, (string SessionId, string Path)?> locate, TimeSpan interval) { try { using var cts = CancellationTokenSource.CreateLinkedTokenSource( agent.ReadCts.Token, agent.ExitedCts.Token, _shutdownCts.Token); - var discovery = new TranscriptDiscovery(_time, SessionIdPollInterval); + var discovery = new TranscriptDiscovery(_time, interval); using var warnCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); var warnTask = WarnIfStillUnlinkedAsync(agent, warnCts.Token); diff --git a/src/Capacitor.Cli.Daemon/Services/IHostedAgentRuntime.cs b/src/Capacitor.Cli.Daemon/Services/IHostedAgentRuntime.cs index 761d405a49..88ca21a94c 100644 --- a/src/Capacitor.Cli.Daemon/Services/IHostedAgentRuntime.cs +++ b/src/Capacitor.Cli.Daemon/Services/IHostedAgentRuntime.cs @@ -58,6 +58,10 @@ internal interface IHostedAgentRuntime : IAsyncDisposable { /// bool EmitsTerminalOutput { get; } + /// True when the agent raises its permission prompts on its own channel and the runtime + /// answers them there, so a prompt its hook also reports to the bridge is not a second one. + bool AsksPermissionsOnStdin => false; + /// /// Terminal byte stream consumed by the orchestrator read loop. PTY runtimes yield real /// terminal output; the ACP runtime yields an empty stream until a follow-up adds a terminal diff --git a/src/Capacitor.Cli.Daemon/Harness/Muse/IMuseMspProcess.cs b/src/Capacitor.Cli.Daemon/Services/IJsonLineProcess.cs similarity index 86% rename from src/Capacitor.Cli.Daemon/Harness/Muse/IMuseMspProcess.cs rename to src/Capacitor.Cli.Daemon/Services/IJsonLineProcess.cs index 1faa7009c4..7e4e52beae 100644 --- a/src/Capacitor.Cli.Daemon/Harness/Muse/IMuseMspProcess.cs +++ b/src/Capacitor.Cli.Daemon/Services/IJsonLineProcess.cs @@ -1,8 +1,8 @@ -namespace Capacitor.Cli.Daemon.Harness.Muse; +namespace Capacitor.Cli.Daemon.Services; /// -/// The long-lived muse serve child speaking MSP JSON-RPC over stdio: one process backs a -/// hosted session and its stdin stays open for that whole lifetime. +/// A long-lived vendor child speaking newline-delimited JSON over stdio: one process backs a hosted +/// session and its stdin stays open for that whole lifetime. /// /// Contracts a real implementation honors: /// 0. may kill a running child, so it is not exit @@ -12,7 +12,7 @@ namespace Capacitor.Cli.Daemon.Harness.Muse; /// 3. calls are serialized, so concurrent writers never tear a /// line on stdin. /// -internal interface IMuseMspProcess : IAsyncDisposable { +internal interface IJsonLineProcess : IAsyncDisposable { int Pid { get; } bool HasExited { get; } diff --git a/src/Capacitor.Cli.Daemon/Harness/Muse/MuseMspProcess.cs b/src/Capacitor.Cli.Daemon/Services/JsonLineProcess.cs similarity index 91% rename from src/Capacitor.Cli.Daemon/Harness/Muse/MuseMspProcess.cs rename to src/Capacitor.Cli.Daemon/Services/JsonLineProcess.cs index 9577873ccf..f97f0ef098 100644 --- a/src/Capacitor.Cli.Daemon/Harness/Muse/MuseMspProcess.cs +++ b/src/Capacitor.Cli.Daemon/Services/JsonLineProcess.cs @@ -1,17 +1,16 @@ using System.Diagnostics; using System.Runtime.CompilerServices; using System.Text; -using Capacitor.Cli.Daemon.Services; using Microsoft.Extensions.Logging; -namespace Capacitor.Cli.Daemon.Harness.Muse; +namespace Capacitor.Cli.Daemon.Services; /// -/// over a real . Stdin stays open for the child's +/// over a real . Stdin stays open for the child's /// lifetime; writes are serialized under , and the pid is captured at /// construction because throws once the object is disposed. /// -internal sealed partial class MuseMspProcess : IMuseMspProcess { +internal sealed partial class JsonLineProcess : IJsonLineProcess { /// Over-cap stderr is dropped, never rotated: the explanation of a failed launch is at the start. internal const int DiagnosticsCap = 4096; @@ -28,12 +27,12 @@ internal sealed partial class MuseMspProcess : IMuseMspProcess { bool IsDisposed => Volatile.Read(ref _disposed) != 0; - internal MuseMspProcess(ProcessStartInfo psi, ILogger logger, TimeProvider time) + internal JsonLineProcess(ProcessStartInfo psi, ILogger logger, TimeProvider time) : this(Process.Start(psi) ?? throw new InvalidOperationException( - $"muse_msp_spawn_failed: '{psi.FileName}' did not start (Process.Start returned null)."), + $"child_spawn_failed: '{psi.FileName}' did not start (Process.Start returned null)."), logger, time) { } - internal MuseMspProcess(Process process, ILogger logger, TimeProvider time) { + internal JsonLineProcess(Process process, ILogger logger, TimeProvider time) { _process = process; _logger = logger; _time = time; @@ -216,6 +215,6 @@ static int SafePid(Process process) { } } - [LoggerMessage(Level = LogLevel.Debug, Message = "Muse MSP stderr: {Length} chars")] + [LoggerMessage(Level = LogLevel.Debug, Message = "Child stderr: {Length} chars")] partial void LogStderrShape(int length); } diff --git a/src/Capacitor.Cli.Daemon/Services/LocalPermissionBridge.cs b/src/Capacitor.Cli.Daemon/Services/LocalPermissionBridge.cs index a7968d52f1..fbb1b82cd0 100644 --- a/src/Capacitor.Cli.Daemon/Services/LocalPermissionBridge.cs +++ b/src/Capacitor.Cli.Daemon/Services/LocalPermissionBridge.cs @@ -15,7 +15,7 @@ namespace Capacitor.Cli.Daemon.Services; internal readonly record struct PermissionAttribution(string? AgentId, string SessionId, string? Cwd); -internal readonly record struct AttributedAgent(string AgentId, PolicySnapshot? PolicySnapshot = null, bool BridgedSinceStart = false); +internal readonly record struct AttributedAgent(string AgentId, PolicySnapshot? PolicySnapshot = null, bool BridgedSinceStart = false, bool AsksOnStdin = false); /// /// Localhost-only HTTP bridge that fronts the server's permission flow for spawned @@ -614,9 +614,7 @@ _ when rawUrl.Equals($"/{submitToken}/flow-message", StringComparison.Ordinal) return; } - var toolName = node["tool_name"]?.GetValue(); - var toolInput = ExtractElement(node, "tool_input"); - var suggestions = ExtractElement(node, "permission_suggestions"); + var toolName = node["tool_name"]?.GetValue(); // HttpListener exposes no per-request "client disconnected" token: every leg below is // bound to the daemon-lifetime token, not the connection. @@ -657,87 +655,16 @@ _ when rawUrl.Equals($"/{submitToken}/flow-message", StringComparison.Ordinal) // Shared (interactive) token: the reserved channel's tool names are not special-cased // here — an interactive session never legitimately carries that server, so an // identically-named tool is untrusted and takes the normal prompt. - var canonicalSessionId = PermissionWire.Canonical(sessionId); - var attributed = canonicalSessionId is null ? null : AttributeHandler?.Invoke(new PermissionAttribution( - node["agent_id"]?.GetValue(), canonicalSessionId, node["cwd"]?.GetValue())); - // Every prompt of a run this daemon launched fresh passes through here, so its refusals are all known. - if (attributed is { BridgedSinceStart: true }) _refusals.MarkBridgedSinceStart(canonicalSessionId!); - var pending = attributed is { } a - ? BuildPending(Guid.NewGuid().ToString("N"), a.AgentId, canonicalSessionId!, vendor, toolName, toolInput, suggestions, - time.GetUtcNow().ToString("O"), ToolUseIdOf(node)) - : null; - - // The launched agent's own policy, and the judge for a call no rule decided, answer - // before a human is asked. Anything the evaluation throws leaves the request on the human lane. - if (vendor is "claude" or "codex" && attributed is { PolicySnapshot: { IsEmpty: false } snapshot } governed) { - ClaudeHostedPolicyResult? policy = null; - try { - var call = new ClaudeHostedPermissionCall(canonicalSessionId!, governed.AgentId, toolName, toolInput, - node["cwd"]?.GetValue(), ToolUseIdOf(node), TranscriptPathOf(node)); - policy = vendor == "codex" - ? await CodexPtyPolicySeam.EvaluateAsync(call, snapshot, time, policyJudge, JudgeBudget, logger, _refusals, ct) - : await ClaudeHostedPolicySeam.EvaluateAsync(call, snapshot, time, policyJudge, judgeState, JudgeBudget, logger, _refusals, ct); - } catch (Exception ex) { - LogPolicyEvaluationFailed(logger, ex, governed.AgentId); - } - - if (policy is { } decided) { - if (decided.Outcome is PolicyOutcome.Allow or PolicyOutcome.Deny) { - var behavior = decided.Outcome == PolicyOutcome.Allow - ? PermissionSettlements.Allow - : PermissionSettlements.Deny; - _decisionLog?.Record(new PermissionDecisionRecord( - time.GetUtcNow().ToString("O"), governed.AgentId, canonicalSessionId!, vendor, - toolName ?? "", behavior, PermissionSettlements.SourcePolicy)); - _ = server.AppendAgentRunEventAsync(governed.AgentId, decided.Event); - // No Register: a call the policy answered raises no card, and the - // decision log plus the run event are its audit trail. - if (behavior == PermissionSettlements.Allow && pending is not null) _broker.OweCall(pending, SubagentIdOf(node)); - RecordRefusal(vendor, governed, canonicalSessionId, node, toolName, toolInput, behavior); - await WriteResponseAsync(context, BuildHookResponseJson(new PermissionDecision(behavior, null, null), vendor), time); - - return; - } - - _ = server.AppendAgentRunEventAsync(governed.AgentId, decided.Event); - } - } - - if (pending is null) { - // Server-only path. - if (attributed is not null) LogPendingOutOfBounds(logger, sessionId); - try { - decision = await server.RequestPermissionAsync(sessionId, toolName, toolInput, suggestions, ct); - } catch (Exception ex) { - LogRequestPermissionFailed(logger, ex, sessionId); - decision = new PermissionDecision("deny", null, null); - } - - RecordRefusal(vendor, attributed, canonicalSessionId, node, toolName, toolInput, decision.Behavior); - } else { - var settlementTask = _broker.Register(pending, SubagentIdOf(node)); - _ = RunServerLegAsync(pending, toolName, toolInput, suggestions, settlementTask, ct); - - PermissionSettlement settlement; - try { - settlement = await settlementTask.WaitAsync(ct); - } catch (OperationCanceledException) { - // Shutdown: claim rather than inspect. Losing means another party settled first. - if (_broker.TrySettle(pending.RequestId, PermissionSettlements.DenyDecision, - PermissionSettlements.Deny, PermissionSettlements.SourceDaemonShutdown)) { - await WriteResponseAsync(context, BuildHookResponseJson(PermissionSettlements.DenyDecision, vendor), time); - return; - } - settlement = await settlementTask; - } + var attributed = Attribute(node, sessionId); - _decisionLog?.Record(new PermissionDecisionRecord( - time.GetUtcNow().ToString("O"), pending.AgentId, pending.SessionId, pending.Vendor, - pending.ToolName, settlement.Outcome, settlement.Source)); - RecordRefusal(vendor, attributed, canonicalSessionId, node, toolName, toolInput, settlement.Decision.Behavior); - await WriteResponseAsync(context, BuildHookResponseJson(settlement.Decision, vendor), time); + // Claude asks such an agent again on stdin whatever the hook answers, so a decision + // here would raise a second card for the same call. An empty answer is no decision. + if (attributed is { AsksOnStdin: true }) { + await WriteResponseAsync(context, "{}", time); return; } + + decision = await DecideAttendedAsync(node, sessionId, vendor, attributed, ct); } await WriteResponseAsync(context, BuildHookResponseJson(decision, vendor), time); @@ -780,6 +707,108 @@ static async Task WriteResponseAsync(HttpListenerContext context, string respons } } + /// Decides a request a hosted agent raised on its own channel instead of through its hook. + /// has the shape the hook posts here. + internal async Task DecideAsync(JsonNode request, string vendor, CancellationToken ct) { + var sessionId = request["session_id"]?.GetValue()?.Replace("-", ""); + if (string.IsNullOrWhiteSpace(sessionId)) return PermissionSettlements.DenyDecision; + + using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, _cts?.Token ?? CancellationToken.None); + return await DecideAttendedAsync(request, sessionId, vendor, Attribute(request, sessionId), linked.Token); + } + + AttributedAgent? Attribute(JsonNode node, string sessionId) => + PermissionWire.Canonical(sessionId) is { } canonical + ? AttributeHandler?.Invoke(new PermissionAttribution( + node["agent_id"]?.GetValue(), canonical, node["cwd"]?.GetValue())) + : null; + + /// The agent's own policy, then the judge, then a human. Cancelling + /// denies a request still waiting on a human. + async Task DecideAttendedAsync( + JsonNode node, string sessionId, string vendor, AttributedAgent? attributed, CancellationToken ct) { + var canonicalSessionId = PermissionWire.Canonical(sessionId); + var toolName = node["tool_name"]?.GetValue(); + var toolInput = ExtractElement(node, "tool_input"); + var suggestions = ExtractElement(node, "permission_suggestions"); + // Every prompt of a run this daemon launched fresh passes through here, so its refusals are all known. + if (attributed is { BridgedSinceStart: true }) _refusals.MarkBridgedSinceStart(canonicalSessionId!); + var pending = attributed is { } a + ? BuildPending(Guid.NewGuid().ToString("N"), a.AgentId, canonicalSessionId!, vendor, toolName, toolInput, suggestions, + time.GetUtcNow().ToString("O"), ToolUseIdOf(node)) + : null; + + // Anything the evaluation throws leaves the request on the human lane. + if (vendor is "claude" or "codex" && attributed is { PolicySnapshot: { IsEmpty: false } snapshot } governed) { + ClaudeHostedPolicyResult? policy = null; + try { + var call = new ClaudeHostedPermissionCall(canonicalSessionId!, governed.AgentId, toolName, toolInput, + node["cwd"]?.GetValue(), ToolUseIdOf(node), TranscriptPathOf(node)); + policy = vendor == "codex" + ? await CodexPtyPolicySeam.EvaluateAsync(call, snapshot, time, policyJudge, JudgeBudget, logger, _refusals, ct) + : await ClaudeHostedPolicySeam.EvaluateAsync(call, snapshot, time, policyJudge, judgeState, JudgeBudget, logger, _refusals, ct); + } catch (Exception ex) { + LogPolicyEvaluationFailed(logger, ex, governed.AgentId); + } + + if (policy is { } decided) { + if (decided.Outcome is PolicyOutcome.Allow or PolicyOutcome.Deny) { + var behavior = decided.Outcome == PolicyOutcome.Allow + ? PermissionSettlements.Allow + : PermissionSettlements.Deny; + _decisionLog?.Record(new PermissionDecisionRecord( + time.GetUtcNow().ToString("O"), governed.AgentId, canonicalSessionId!, vendor, + toolName ?? "", behavior, PermissionSettlements.SourcePolicy)); + _ = server.AppendAgentRunEventAsync(governed.AgentId, decided.Event); + // No Register: a call the policy answered raises no card, and the + // decision log plus the run event are its audit trail. + if (behavior == PermissionSettlements.Allow && pending is not null) _broker.OweCall(pending, SubagentIdOf(node)); + RecordRefusal(vendor, governed, canonicalSessionId, node, toolName, toolInput, behavior); + + return new PermissionDecision(behavior, null, null); + } + + _ = server.AppendAgentRunEventAsync(governed.AgentId, decided.Event); + } + } + + if (pending is null) { + if (attributed is not null) LogPendingOutOfBounds(logger, sessionId); + PermissionDecision decision; + try { + decision = await server.RequestPermissionAsync(sessionId, toolName, toolInput, suggestions, ct); + } catch (Exception ex) { + LogRequestPermissionFailed(logger, ex, sessionId); + decision = new PermissionDecision("deny", null, null); + } + + RecordRefusal(vendor, attributed, canonicalSessionId, node, toolName, toolInput, decision.Behavior); + return decision; + } + + var settlementTask = _broker.Register(pending, SubagentIdOf(node)); + _ = RunServerLegAsync(pending, toolName, toolInput, suggestions, settlementTask, ct); + + PermissionSettlement settlement; + try { + settlement = await settlementTask.WaitAsync(ct); + } catch (OperationCanceledException) { + // Claim rather than inspect. Losing means another party settled first. + var (outcome, source) = _cts is { IsCancellationRequested: true } + ? (PermissionSettlements.Deny, PermissionSettlements.SourceDaemonShutdown) + : (PermissionSettlements.Withdrawn, PermissionSettlements.SourceAgentGone); + if (_broker.TrySettle(pending.RequestId, PermissionSettlements.DenyDecision, outcome, source)) + return PermissionSettlements.DenyDecision; + settlement = await settlementTask; + } + + _decisionLog?.Record(new PermissionDecisionRecord( + time.GetUtcNow().ToString("O"), pending.AgentId, pending.SessionId, pending.Vendor, + pending.ToolName, settlement.Outcome, settlement.Source)); + RecordRefusal(vendor, attributed, canonicalSessionId, node, toolName, toolInput, settlement.Decision.Behavior); + return settlement.Decision; + } + internal static PermissionPendingDto? BuildPending( string requestId, string agentId, string sessionId, string vendor, string? toolName, JsonElement? toolInput, JsonElement? suggestions, string requestedAt, string? toolUseId = null) { diff --git a/src/Capacitor.Cli.Daemon/Services/PtyHostedAgentRuntimeFactory.cs b/src/Capacitor.Cli.Daemon/Services/PtyHostedAgentRuntimeFactory.cs index 43001eec6b..f934c94e76 100644 --- a/src/Capacitor.Cli.Daemon/Services/PtyHostedAgentRuntimeFactory.cs +++ b/src/Capacitor.Cli.Daemon/Services/PtyHostedAgentRuntimeFactory.cs @@ -47,6 +47,41 @@ TimeProvider time /// aborts an in-flight PTY launch. /// public async Task StartAsync(RuntimeStartContext ctx, CancellationToken ct) { + var launcherCtx = await PrepareAsync(launcher, ctx, ex => LogPrepareSoftFailure(ex, ctx.AgentId)); + + var launchArgs = launcher.BuildArgs(launcherCtx); + + string? launchPrompt = null; + if (launcherCtx.Prompt is { Length: > 0 } prompt + && launchArgs.Args.Contains(prompt) + && !CommandLinePrompt.StaysInline(prompt, launchArgs.Args)) { + launcherCtx = launcherCtx with { Prompt = null }; + launchArgs = launcher.BuildArgs(launcherCtx); + launchPrompt = prompt; + LogPromptPastedAfterSpawn(ctx.AgentId, prompt.Length); + } + + var env = HostedEnv(ctx); + + // Read before the spawn so a throw cannot orphan the PTY. Gates the multi-CR submit spray; see + // PtyHostedAgentRuntime.SubmitAsync. + var disablesApprovalPrompts = launcher.DisablesApprovalPrompts(launcherCtx); + + var pty = ptyFactory.Spawn(launcher.CliPath, launchArgs.Args, ctx.Worktree.Path, env, ctx.Cols, ctx.Rows); + var runtime = new PtyHostedAgentRuntime(ctx.Vendor, pty, time, disablesApprovalPrompts, logger); + + if (launchPrompt is not null) { + runtime.DeliverLaunchPrompt(launchPrompt); + _ = LogLaunchPromptOutcomeAsync(runtime.LaunchPromptDelivery!, ctx.AgentId); + } + + return new HostedRuntimeStart(runtime, launchArgs.McpConfigPath); + } + + /// Builds the launcher's context and runs its . A + /// failure the launch can survive goes to and the launch carries on. + internal static async Task PrepareAsync( + IHostedAgentLauncher launcher, RuntimeStartContext ctx, Action softFailure) { var launcherCtx = new LauncherContext( AgentId: ctx.AgentId, SourceRepoPath: ctx.SourceRepoPath, @@ -83,26 +118,16 @@ public async Task StartAsync(RuntimeStartContext ctx, Cancel // Propagate — the orchestrator maps this to LaunchFailed + worktree cleanup. throw; } catch (Exception ex) { - LogPrepareSoftFailure(ex, ctx.AgentId); + softFailure(ex); } finally { if (owned) WorktreeManager.RecordDaemonBaseline(launcherCtx.Worktree.Path, before); } - var launchArgs = launcher.BuildArgs(launcherCtx); - - string? launchPrompt = null; - if (launcherCtx.Prompt is { Length: > 0 } prompt - && launchArgs.Args.Contains(prompt) - && !CommandLinePrompt.StaysInline(prompt, launchArgs.Args)) { - launcherCtx = launcherCtx with { Prompt = null }; - launchArgs = launcher.BuildArgs(launcherCtx); - launchPrompt = prompt; - LogPromptPastedAfterSpawn(ctx.AgentId, prompt.Length); - } - - var args = launchArgs.Args; - var mcpConfigPath = launchArgs.McpConfigPath; + return launcherCtx; + } + /// What the daemon exports into a hosted agent on top of what it inherits. + internal static Dictionary HostedEnv(RuntimeStartContext ctx) { var env = new Dictionary(); // A private agent gets what a --private local spawn gets: nothing that ties its session to @@ -113,10 +138,8 @@ public async Task StartAsync(RuntimeStartContext ctx, Cancel HostedLaunchMarkers.ApplyOverlay(env, ctx); } - // Phase B (D4 §6.4(3)): stamp the daemon-identity markers so a restarted daemon's - // OrphanReaper env-marker scan can reap a recordless survivor of a PRIOR incarnation of THIS - // daemon (same KCAP_DAEMON_ID, older KCAP_DAEMON_EPOCH) — and never touch another daemon's - // child or a live agent of the current incarnation. + // A restarted daemon reaps a recordless survivor only when these name it (same id, older + // epoch), so another daemon's child and a live agent of this incarnation are never touched. if (!string.IsNullOrEmpty(ctx.DaemonId)) env["KCAP_DAEMON_ID"] = ctx.DaemonId; if (!string.IsNullOrEmpty(ctx.DaemonEpoch)) env["KCAP_DAEMON_EPOCH"] = ctx.DaemonEpoch; @@ -136,19 +159,7 @@ public async Task StartAsync(RuntimeStartContext ctx, Cancel env["KCAP_REVIEW_PR"] = reviewEnv.PrNumber.ToString(); } - // Read before the spawn so a throw cannot orphan the PTY. Gates the multi-CR submit spray; see - // PtyHostedAgentRuntime.SubmitAsync. - var disablesApprovalPrompts = launcher.DisablesApprovalPrompts(launcherCtx); - - var pty = ptyFactory.Spawn(launcher.CliPath, args, ctx.Worktree.Path, env, ctx.Cols, ctx.Rows); - var runtime = new PtyHostedAgentRuntime(ctx.Vendor, pty, time, disablesApprovalPrompts, logger); - - if (launchPrompt is not null) { - runtime.DeliverLaunchPrompt(launchPrompt); - _ = LogLaunchPromptOutcomeAsync(runtime.LaunchPromptDelivery!, ctx.AgentId); - } - - return new HostedRuntimeStart(runtime, mcpConfigPath); + return env; } [LoggerMessage(Level = LogLevel.Warning, Message = "Launcher Prepare soft-failure for agent {AgentId} (continuing)")] diff --git a/test/Capacitor.App.Tests.Unit/HomeViewModelTests.cs b/test/Capacitor.App.Tests.Unit/HomeViewModelTests.cs index 38bb6c1374..29b4e30ac5 100644 --- a/test/Capacitor.App.Tests.Unit/HomeViewModelTests.cs +++ b/test/Capacitor.App.Tests.Unit/HomeViewModelTests.cs @@ -1832,7 +1832,7 @@ await AvaloniaSession.WithImmediateRxScheduler(async () => { var opened = new List<(string AgentId, int Generation)>(); using var vm = new HomeViewModel( daemon, new AppStateStore(path), launch, Known(), TimeProvider.System, - openSessionIfCurrent: (id, generation) => opened.Add((id, generation)), + openSessionIfCurrent: (id, generation, _) => opened.Add((id, generation)), daemons: remote.Daemons, viewerId: _ => Task.FromResult("u1"), laneStatus: lane.Status); await vm.SelectMachineAsync("home-pc", isLocal: false); @@ -1853,7 +1853,7 @@ await AvaloniaSession.WithImmediateRxScheduler(async () => { var opened = new List<(string AgentId, int Generation)>(); using var vm = new HomeViewModel( daemon, new AppStateStore(path), launch, Known(), TimeProvider.System, - openSessionIfCurrent: (id, generation) => opened.Add((id, generation))); + openSessionIfCurrent: (id, generation, _) => opened.Add((id, generation))); await vm.SelectRepositoryAsync("/repo/a"); await vm.StartCommand.Execute(); diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/ClaudeHostedAgentRuntimeFactoryTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/ClaudeHostedAgentRuntimeFactoryTests.cs new file mode 100644 index 0000000000..6fc638eea0 --- /dev/null +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/ClaudeHostedAgentRuntimeFactoryTests.cs @@ -0,0 +1,89 @@ +using Capacitor.Cli.Core.LocalIpc; +using Capacitor.Cli.Daemon.Harness.Claude; +using Capacitor.Cli.Daemon.Services; +using Capacitor.Cli.Daemon.Tests.Unit.Services; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Capacitor.Cli.Daemon.Tests.Unit.Harness.Claude; + +/// Which Claude launches leave the PTY: only an interactive, owned-worktree, non-private +/// launch on a daemon set to stream-json, and only when a bridge can answer its prompts. +public class ClaudeHostedAgentRuntimeFactoryTests { + [TempHome] public required TempHome Home { get; init; } + + sealed class StubPtyFactory : IHostedAgentRuntimeFactory { + public string CliPath => "claude"; + public string Vendor => "claude"; + public bool IsAvailable() => true; + public bool SupportsUnattended => true; + public Task StartAsync(RuntimeStartContext ctx, CancellationToken ct) => + throw new NotSupportedException(); + } + + sealed class StubLauncher : IHostedAgentLauncher { + public string Vendor => "claude"; + public string CliPath => "claude"; + public bool SupportsUnattended => true; + public bool IsAvailable() => true; + public void Prepare(LauncherContext ctx) { } + public LaunchArgs BuildArgs(LauncherContext ctx) => new([], null); + public LaunchArgs BuildPassthrough(LauncherContext ctx, IReadOnlyList userArgs) => new([], null); + public void Cleanup(AgentInstance agent) { } + } + + ClaudeHostedAgentRuntimeFactory Factory( + string transport, Func? processSource = null) => + new(new StubLauncher(), new StubPtyFactory(), new DaemonConfig { ClaudeTransport = transport }, + new LocalPermissionBridge(new FakeServerConnection(respond: null), NullLogger.Instance, + EphemeralLoopbackPortSource.Instance, TimeProvider.System), + TestHarnesses.Under(Home).Of().Paths, + NullLoggerFactory.Instance, TimeProvider.System, processSource); + + static RuntimeStartContext Ctx( + bool isReview = false, bool isReviewFlow = false, bool isPrivate = false, + WorkLocation work = WorkLocation.OwnedWorktree, string? bridge = "http://127.0.0.1:1/t", + string worktree = "/wt") => new( + AgentId: "agent-1", Vendor: "claude", SourceRepoPath: worktree, + Worktree: new WorktreeInfo(Path: worktree, Branch: "wt", SourceRepo: worktree), + Prompt: "hi", Model: null, Effort: null, Tools: null, + IsReview: isReview, IsReviewFlow: isReviewFlow, Review: null, + Cols: 80, Rows: 24, ServerUrl: "https://t.example", DaemonBridgeUrl: bridge, CapacitorPath: "/opt/kcap", + Work: work, IsPrivate: isPrivate); + + [Test] + [Arguments("stream-json", false, false, false, false, true, true)] + [Arguments("pty", false, false, false, false, true, false)] + [Arguments("anything-else", false, false, false, false, true, false)] + [Arguments("stream-json", true, false, false, false, true, false)] + [Arguments("stream-json", false, true, false, false, true, false)] + [Arguments("stream-json", false, false, true, false, true, false)] + [Arguments("stream-json", false, false, false, true, true, false)] + [Arguments("stream-json", false, false, false, false, false, false)] + public async Task Routes_only_bridged_interactive_owned_launches_to_stream_json( + string transport, bool review, bool reviewFlow, bool isPrivate, bool borrowed, bool bridged, bool expected) { + var ctx = Ctx(review, reviewFlow, isPrivate, + borrowed ? WorkLocation.BorrowedCwd : WorkLocation.OwnedWorktree, bridged ? "http://127.0.0.1:1/t" : null); + + await Assert.That(Factory(transport).UsesStreamJson(ctx)).IsEqualTo(expected); + } + + /// The prompt never reaches the command line: it is the first message after the + /// handshake. + [Test] + public async Task The_prompt_is_the_first_stdin_message_and_never_an_argument() { + using var wt = new TempDir(); + var process = new FakeStreamJsonProcess(); + System.Diagnostics.ProcessStartInfo? spawned = null; + var factory = Factory("stream-json", psi => { spawned = psi; return process; }); + + var start = await factory.StartAsync(Ctx(worktree: wt.Path), CancellationToken.None); + await using var _ = start.Runtime; + + await Assert.That(spawned!.ArgumentList).DoesNotContain("hi"); + await Assert.That(spawned.ArgumentList).Contains("--permission-prompt-tool"); + await Assert.That(spawned.WorkingDirectory).IsEqualTo(wt.Path); + await Assert.That(spawned.Environment[Core.HostedAgent.AgentIdVar]).IsEqualTo("agent-1"); + await Assert.That(await process.NextWriteAsync()).Contains("\"initialize\""); + await Assert.That(await process.NextWriteAsync()).Contains("\"content\":\"hi\""); + } +} diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/ClaudeStreamJsonHostedAgentRuntimeTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/ClaudeStreamJsonHostedAgentRuntimeTests.cs new file mode 100644 index 0000000000..56527dddfa --- /dev/null +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/ClaudeStreamJsonHostedAgentRuntimeTests.cs @@ -0,0 +1,141 @@ +using System.Text.Json; +using System.Text.Json.Nodes; +using Capacitor.Cli.Core; +using Capacitor.Cli.Daemon.Harness.Claude; +using Capacitor.Cli.Daemon.Services; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Capacitor.Cli.Daemon.Tests.Unit.Harness.Claude; + +/// The stdio half of a stream-json Claude agent: user input and interrupts go out as JSON +/// lines, and a can_use_tool request is decided through the bridge and answered on stdin. +public class ClaudeStreamJsonHostedAgentRuntimeTests { + const string Session = "6ba7b810-9dad-11d1-80b4-00c04fd430c8"; + + sealed class Rig : IAsyncDisposable { + public FakeStreamJsonProcess Process { get; } = new(); + public List Asked { get; } = []; + public TaskCompletionSource Decision { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource AskToken { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public ClaudeStreamJsonHostedAgentRuntime Runtime { get; } + + public Rig() => + Runtime = new ClaudeStreamJsonHostedAgentRuntime( + Process, + (request, ct) => { + lock (Asked) Asked.Add(request); + AskToken.TrySetResult(ct); + return Decision.Task.WaitAsync(ct); + }, + session => $"/projects/{session}.jsonl", + "agent-1", "/wt", NullLogger.Instance, TimeProvider.System); + + public void Init() => Process.Push($$"""{"type":"system","subtype":"init","session_id":"{{Session}}"}"""); + + public void Ask(string requestId, string toolName, string input) => Process.Push( + $$$"""{"type":"control_request","request_id":"{{{requestId}}}","request":{"subtype":"can_use_tool","tool_name":"{{{toolName}}}","input":{{{input}}},"permission_suggestions":[{"type":"setMode","mode":"acceptEdits","destination":"session"}],"tool_use_id":"toolu_01X"}}"""); + + public ValueTask DisposeAsync() => Runtime.DisposeAsync(); + } + + static JsonElement Parse(string json) => JsonDocument.Parse(json).RootElement; + + [Test] + public async Task A_permission_request_is_decided_through_the_bridge_in_the_hooks_shape() { + await using var rig = new Rig(); + rig.Init(); + rig.Ask("req-1", "Bash", """{"command":"touch x"}"""); + await rig.AskToken.Task.WaitAsync(TimeSpan.FromSeconds(10), TimeProvider.System); + + var asked = rig.Asked.Single(); + await Assert.That(asked["session_id"]!.GetValue()).IsEqualTo(Session); + await Assert.That(asked["tool_name"]!.GetValue()).IsEqualTo("Bash"); + await Assert.That(asked["tool_input"]!["command"]!.GetValue()).IsEqualTo("touch x"); + await Assert.That(asked["permission_suggestions"]!.AsArray().Count).IsEqualTo(1); + await Assert.That(asked["agent_id"]!.GetValue()).IsEqualTo("agent-1"); + await Assert.That(asked["cwd"]!.GetValue()).IsEqualTo("/wt"); + await Assert.That(asked["tool_use_id"]!.GetValue()).IsEqualTo("toolu_01X"); + await Assert.That(asked["transcript_path"]!.GetValue()).IsEqualTo($"/projects/{Session}.jsonl"); + } + + /// An allow without an edited input hands the call's own input back: Claude validates + /// updatedInput as the whole tool input. + [Test] + public async Task An_allow_answers_with_the_input_and_a_deny_with_a_message() { + await using var rig = new Rig(); + rig.Init(); + rig.Ask("req-1", "Bash", """{"command":"touch x"}"""); + rig.Decision.SetResult(new PermissionDecision("allow", null, null)); + + var response = Parse(await rig.Process.NextWriteAsync()).GetProperty("response"); + await Assert.That(response.GetProperty("request_id").GetString()).IsEqualTo("req-1"); + await Assert.That(response.GetProperty("response").GetProperty("behavior").GetString()).IsEqualTo("allow"); + await Assert.That(response.GetProperty("response").GetProperty("updatedInput").GetProperty("command").GetString()).IsEqualTo("touch x"); + + var deny = ClaudeStreamJsonHostedAgentRuntime.PermissionResponse("req-2", PermissionSettlements.DenyDecision, null); + await Assert.That(deny["response"]!["response"]!["behavior"]!.GetValue()).IsEqualTo("deny"); + await Assert.That(deny["response"]!["response"]!["message"]).IsNotNull(); + } + + /// A question's answers come back from the server as the whole updated input. + [Test] + public async Task A_decisions_updated_input_and_rules_replace_the_calls_own() { + var answered = Parse("""{"questions":[{"question":"Colour?"}],"answers":{"Colour?":"Purple"}}"""); + var rules = Parse("""[{"type":"setMode","mode":"acceptEdits","destination":"session"}]"""); + + var response = ClaudeStreamJsonHostedAgentRuntime.PermissionResponse( + "req-1", new PermissionDecision("allow", rules, answered), Parse("""{"questions":[]}"""))["response"]!["response"]!; + + await Assert.That(response["updatedInput"]!["answers"]!["Colour?"]!.GetValue()).IsEqualTo("Purple"); + await Assert.That(response["updatedPermissions"]!.AsArray().Count).IsEqualTo(1); + } + + [Test] + public async Task A_withdrawn_request_cancels_its_decision_and_is_not_answered() { + await using var rig = new Rig(); + rig.Init(); + rig.Ask("req-1", "Bash", """{"command":"touch x"}"""); + var token = await rig.AskToken.Task.WaitAsync(TimeSpan.FromSeconds(10), TimeProvider.System); + + rig.Process.Push("""{"type":"control_cancel_request","request_id":"req-1"}"""); + + await Assert.That(async () => await Task.Delay(Timeout.Infinite, token)).Throws(); + await Task.Delay(50); + await Assert.That(rig.Process.HasPendingWrite).IsFalse(); + } + + /// Nothing can be attributed before the session is known, so nothing is asked. + [Test] + public async Task A_request_before_the_session_is_known_is_denied_without_asking() { + await using var rig = new Rig(); + rig.Ask("req-1", "Bash", """{"command":"touch x"}"""); + + var response = Parse(await rig.Process.NextWriteAsync()).GetProperty("response").GetProperty("response"); + + await Assert.That(response.GetProperty("behavior").GetString()).IsEqualTo("deny"); + await Assert.That(rig.Asked.Count).IsEqualTo(0); + } + + [Test] + public async Task Input_is_a_user_message_and_escape_is_an_interrupt() { + await using var rig = new Rig(); + + await rig.Runtime.SendUserInputAsync("fix the login bug"); + var user = Parse(await rig.Process.NextWriteAsync()); + await Assert.That(user.GetProperty("type").GetString()).IsEqualTo("user"); + await Assert.That(user.GetProperty("message").GetProperty("content").GetString()).IsEqualTo("fix the login bug"); + + await rig.Runtime.SendSpecialKeyAsync("Escape"); + var interrupt = Parse(await rig.Process.NextWriteAsync()); + await Assert.That(interrupt.GetProperty("request").GetProperty("subtype").GetString()).IsEqualTo("interrupt"); + } + + [Test] + public async Task Input_is_refused_once_the_agent_has_exited() { + await using var rig = new Rig(); + rig.Process.EndOfStream(); + await foreach (var _ in rig.Runtime.ReadOutputAsync()) { } + + await Assert.That(() => rig.Runtime.SendUserInputAsync("late")).Throws(); + } +} diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/FakeStreamJsonProcess.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/FakeStreamJsonProcess.cs new file mode 100644 index 0000000000..40ac1f6704 --- /dev/null +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Claude/FakeStreamJsonProcess.cs @@ -0,0 +1,55 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Capacitor.Cli.Daemon.Services; + +namespace Capacitor.Cli.Daemon.Tests.Unit.Harness.Claude; + +/// A claude -p child: writes a line to its stdout, +/// reads what reached its stdin, and is it exiting. +internal sealed class FakeStreamJsonProcess : IJsonLineProcess { + readonly Channel _lines = Channel.CreateUnbounded(); + readonly Channel _writes = Channel.CreateUnbounded(); + + public int Pid => 4343; + public bool HasExited { get; private set; } + public int? ExitCode { get; private set; } + public string? Diagnostics => null; + + public void Push(string line) => _lines.Writer.TryWrite(line); + + public void EndOfStream() { + HasExited = true; + ExitCode = 0; + _lines.Writer.TryComplete(); + } + + /// The next line written to stdin. + public Task NextWriteAsync() => _writes.Reader.ReadAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(10), TimeProvider.System); + + public bool HasPendingWrite => _writes.Reader.TryPeek(out _); + + public async IAsyncEnumerable ReadLinesAsync([EnumeratorCancellation] CancellationToken ct) { + while (await _lines.Reader.WaitToReadAsync(ct).ConfigureAwait(false)) + while (_lines.Reader.TryRead(out var line)) + yield return line; + } + + public Task WriteLineAsync(string json, CancellationToken ct) { + _writes.Writer.TryWrite(json); + return Task.CompletedTask; + } + + public Task WaitForExitAsync(TimeSpan? timeout = null) => Task.CompletedTask; + + public Task CloseInputAsync(TimeSpan timeout) => Task.CompletedTask; + + public Task TerminateAsync(TimeSpan? timeout = null) { + EndOfStream(); + return Task.CompletedTask; + } + + public ValueTask DisposeAsync() { + _lines.Writer.TryComplete(); + return ValueTask.CompletedTask; + } +} diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/FakeMuseMspProcess.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/FakeMuseMspProcess.cs index 1f87b0a7a7..ed7ee54599 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/FakeMuseMspProcess.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/FakeMuseMspProcess.cs @@ -1,15 +1,16 @@ using System.Runtime.CompilerServices; using System.Threading.Channels; using Capacitor.Cli.Daemon.Harness.Muse; +using Capacitor.Cli.Daemon.Services; namespace Capacitor.Cli.Daemon.Tests.Unit.Harness.Muse; /// -/// Interactive fake: puts a line on the child's +/// Interactive fake: puts a line on the child's /// stdout at any moment, captures what the runtime wrote to stdin, and /// is the child exiting. /// -internal sealed class FakeMuseMspProcess : IMuseMspProcess { +internal sealed class FakeMuseMspProcess : IJsonLineProcess { readonly Channel _lines = Channel.CreateUnbounded(); readonly Lock _gate = new(); readonly List _writes = []; diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/MuseHostedAgentRuntimeFactoryTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/MuseHostedAgentRuntimeFactoryTests.cs index 56dc9ae61c..01e9f802cc 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/MuseHostedAgentRuntimeFactoryTests.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/MuseHostedAgentRuntimeFactoryTests.cs @@ -95,7 +95,7 @@ Rig NewRig( processSource: (psi, _) => { rig!.Spawned.Add(psi); rig.MarkerFilesAtSpawn.Add(Directory.Exists(MarkerDir) ? Directory.GetFiles(MarkerDir) : []); - return Task.FromResult(host.Process); + return Task.FromResult(host.Process); }, binaryExists: _ => binary, resolveVersion: _ => installed, diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/MuseMspProcessTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/MuseMspProcessTests.cs deleted file mode 100644 index c7b45c33b4..0000000000 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Harness/Muse/MuseMspProcessTests.cs +++ /dev/null @@ -1,28 +0,0 @@ -using System.Text; -using Capacitor.Cli.Daemon.Harness.Muse; - -namespace Capacitor.Cli.Daemon.Tests.Unit.Harness.Muse; - -public class MuseMspProcessTests { - [Test] - public async Task One_line_longer_than_the_cap_is_cut_at_it() { - var diagnostics = new StringBuilder(); - - MuseMspProcess.AppendCapped(diagnostics, new string('x', MuseMspProcess.DiagnosticsCap * 3)); - - await Assert.That(diagnostics.Length).IsEqualTo(MuseMspProcess.DiagnosticsCap); - } - - [Test] - public async Task Lines_after_the_cap_is_reached_are_dropped() { - var diagnostics = new StringBuilder(); - - MuseMspProcess.AppendCapped(diagnostics, "first"); - MuseMspProcess.AppendCapped(diagnostics, new string('x', MuseMspProcess.DiagnosticsCap)); - MuseMspProcess.AppendCapped(diagnostics, "late"); - - await Assert.That(diagnostics.Length).IsEqualTo(MuseMspProcess.DiagnosticsCap); - await Assert.That(diagnostics.ToString()).StartsWith("first\n"); - await Assert.That(diagnostics.ToString()).DoesNotContain("late"); - } -} diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/JsonLineProcessTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/JsonLineProcessTests.cs new file mode 100644 index 0000000000..2adb2c0a99 --- /dev/null +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/JsonLineProcessTests.cs @@ -0,0 +1,28 @@ +using System.Text; +using Capacitor.Cli.Daemon.Services; + +namespace Capacitor.Cli.Daemon.Tests.Unit.Services; + +public class JsonLineProcessTests { + [Test] + public async Task One_line_longer_than_the_cap_is_cut_at_it() { + var diagnostics = new StringBuilder(); + + JsonLineProcess.AppendCapped(diagnostics, new string('x', JsonLineProcess.DiagnosticsCap * 3)); + + await Assert.That(diagnostics.Length).IsEqualTo(JsonLineProcess.DiagnosticsCap); + } + + [Test] + public async Task Lines_after_the_cap_is_reached_are_dropped() { + var diagnostics = new StringBuilder(); + + JsonLineProcess.AppendCapped(diagnostics, "first"); + JsonLineProcess.AppendCapped(diagnostics, new string('x', JsonLineProcess.DiagnosticsCap)); + JsonLineProcess.AppendCapped(diagnostics, "late"); + + await Assert.That(diagnostics.Length).IsEqualTo(JsonLineProcess.DiagnosticsCap); + await Assert.That(diagnostics.ToString()).StartsWith("first\n"); + await Assert.That(diagnostics.ToString()).DoesNotContain("late"); + } +} diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/LocalPermissionBridgeInteractiveTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/LocalPermissionBridgeInteractiveTests.cs index 1b493a1b90..d7cf803db1 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/LocalPermissionBridgeInteractiveTests.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/LocalPermissionBridgeInteractiveTests.cs @@ -20,10 +20,10 @@ sealed class Harness : IAsyncDisposable { public LocalPermissionBridge Bridge { get; } public HttpClient Client { get; } = new() { Timeout = TimeSpan.FromSeconds(30) }; - public Harness(string? attributeTo = "agent-1") { + public Harness(string? attributeTo = "agent-1", bool asksOnStdin = false) { Log = new PermissionDecisionLog(Tmp.Path, NullLogger.Instance); Bridge = new LocalPermissionBridge(Server, NullLogger.Instance, EphemeralLoopbackPortSource.Instance, TimeProvider.System, Broker, Log) { - AttributeHandler = attributeTo is null ? _ => null : _ => new AttributedAgent(attributeTo), + AttributeHandler = attributeTo is null ? _ => null : _ => new AttributedAgent(attributeTo, AsksOnStdin: asksOnStdin), }; } @@ -320,6 +320,61 @@ public async Task Build_pending_bounds() { await Assert.That(LocalPermissionBridge.BuildPending("r", "a1", Session, "claude", "Bash", null, null, "t", new string('i', PermissionWire.MaxToolUseIdBytes + 1))!.ToolUseId).IsNull(); } + /// Claude asks a stdin agent again on stdin whatever its hook is told, so the hook must get no + /// decision and raise no card of its own. + [Test, NotInParallel(nameof(LocalPermissionBridgeInteractiveTests))] + public async Task A_hook_from_an_agent_that_asks_on_stdin_gets_no_decision_and_raises_nothing() { + await using var h = new Harness(asksOnStdin: true); + await h.StartAsync(); + + using var response = await h.PostAsync(); + + await Assert.That(await response.Content.ReadAsStringAsync()).IsEqualTo("{}"); + await Assert.That(h.Broker.PendingSnapshot().Count).IsEqualTo(0); + await Assert.That(h.Server.Responds.Count).IsEqualTo(0); + } + + [Test, NotInParallel(nameof(LocalPermissionBridgeInteractiveTests))] + public async Task An_in_process_request_takes_the_brokered_path_and_returns_the_settlement() { + await using var h = new Harness(asksOnStdin: true); + h.Server.AwaitScript = (_, ct) => new TaskCompletionSource().Task.WaitAsync(ct); + await h.StartAsync(); + + var decision = h.Bridge.DecideAsync(Request(), "claude", CancellationToken.None); + await WaitUntil(() => h.Broker.PendingSnapshot().Count == 1, "the request is pending"); + var pending = h.Broker.PendingSnapshot().Single(); + await Assert.That(pending.ToolUseId).IsEqualTo("toolu_01X"); + + await Assert.That(h.Broker.TrySettle(pending.RequestId, Allow, "allow", "app")).IsTrue(); + + await Assert.That((await decision).Behavior).IsEqualTo("allow"); + } + + /// The agent withdrawing its request settles the card rather than leaving it for a human. + [Test, NotInParallel(nameof(LocalPermissionBridgeInteractiveTests))] + public async Task Cancelling_an_in_process_request_withdraws_it() { + await using var h = new Harness(asksOnStdin: true); + h.Server.AwaitScript = (_, ct) => new TaskCompletionSource().Task.WaitAsync(ct); + await h.StartAsync(); + using var cts = new CancellationTokenSource(); + + var decision = h.Bridge.DecideAsync(Request(), "claude", cts.Token); + await WaitUntil(() => h.Broker.PendingSnapshot().Count == 1, "the request is pending"); + await cts.CancelAsync(); + + await Assert.That((await decision).Behavior).IsEqualTo("deny"); + await WaitUntil(() => h.Broker.PendingSnapshot().Count == 0, "the request is withdrawn"); + } + + static System.Text.Json.Nodes.JsonObject Request() => new() { + ["session_id"] = Session, + ["tool_name"] = "Bash", + ["tool_input"] = new System.Text.Json.Nodes.JsonObject { ["command"] = "ls" }, + ["tool_use_id"] = "toolu_01X", + ["agent_id"] = "agent-1", + ["cwd"] = "/repo" + }; + static async Task WaitUntil(Func condition, string what) { var deadline = DateTime.UtcNow.AddSeconds(30); while (!condition()) {