Repository navigation
Hand off an ended session from the desktop app #2205
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. Weβll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
73db666
d2edd15
899141b
0b0cbe6
89c90f2
1aaa104
c7b18c0
dec7b8f
335fe3c
9579966
2ec9fa0
9c9c858
e86ee65
e948f14
c9687db
cfea4ed
b589fd5
8ccdc21
ed12cc9
bd35077
f6538c9
3426e11
46c8ad7
dd9a355
b8350a8
560f32e
1fd0756
80681c6
e14165c
9b8f730
d92bd9b
87a7f02
20fa093
57776e3
bd0ab9d
fd2241f
2ff0244
025a00b
9004f0f
576c8fb
03e47c0
20db7c3
268d9d1
1291e48
3a68355
6550d99
9029b67
4a06bb5
03b15db
90bb081
aba5d0b
8b86f68
9e03daf
99bace6
f0b3b66
1dd8ef9
bc98b1d
4773170
93504a8
74f1ce6
40e0a8c
d8a03b7
18ced6a
5a5ec3c
0ede723
35720cf
ccbfb4f
e65cf5c
f7c1fe7
1bea4be
49bba59
cfe5643
2a16403
fe4c877
786438d
ff76a2b
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| using Capacitor.Remote.Models; | ||
|
|
||
| namespace Capacitor.App.Services; | ||
|
|
||
| /// Unavailable is a session the server hides from the caller or does not hold; Context is set only on Ok. | ||
| public sealed record HandoffContextFetch(ServerReadOutcome Outcome, HandoffContextDto? Context); |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| using Capacitor.App.ViewModels; | ||
| using Capacitor.Cli.Core; | ||
|
|
||
| namespace Capacitor.App.Services; | ||
|
|
||
| /// A hand-off composer's contents as the chips showed them: PermissionMode and CodexPosture are the | ||
| /// chip values, not what the launch sent. Error is the failure that handed the draft back. | ||
| public sealed record HandoffDraft( | ||
| string Text, IReadOnlyList<StagedAttachment> Attachments, string Vendor, string Model, string? Effort, | ||
| string? PermissionMode, CodexLaunchPosture? CodexPosture, string? Error); |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| using System.Reactive.Disposables; | ||
| using System.Reactive.Disposables.Fluent; | ||
| using System.Reactive.Linq; | ||
| using System.Reactive.Subjects; | ||
| using Capacitor.App.ViewModels; | ||
| using DynamicData; | ||
| using ReactiveUI.Reactive; | ||
|
|
||
| namespace Capacitor.App.Services; | ||
|
|
||
| /// Hand-off drafts awaiting their launch's verdict, for the app's lifetime: the composer that sent | ||
| /// one is gone once the new session opens, and a failure must hand the draft back to the source | ||
| /// session's next composer. Failures are buffered from construction, because one can arrive before | ||
| /// the launch call returns and the hold that would match it is made. | ||
| public sealed class HandoffDrafts : IDisposable { | ||
| static readonly TimeSpan PendingTtl = TimeSpan.FromMinutes(10); | ||
| static readonly TimeSpan RecentFailureTtl = TimeSpan.FromSeconds(30); | ||
|
|
||
| sealed record Held(string SourceSessionId, HandoffDraft Draft, DateTimeOffset At, Action<string>? OnFailed); | ||
|
|
||
| readonly Lock _lock = new(); | ||
| readonly Dictionary<string, Held> _pending = new(StringComparer.Ordinal); | ||
| readonly Dictionary<string, (string Reason, DateTimeOffset At)> _recentFailures = new(StringComparer.Ordinal); | ||
| readonly Dictionary<string, HandoffDraft> _failed = new(StringComparer.Ordinal); | ||
| readonly IAgentDirectory? _directory; | ||
| readonly TimeProvider _time; | ||
| readonly CompositeDisposable _subscriptions = new(); | ||
| readonly Subject<string> _failedFor = new(); | ||
|
|
||
| public HandoffDrafts(IObservable<LaunchFailure> failures, IAgentDirectory? directory, TimeProvider time) { | ||
| _directory = directory; | ||
| _time = time; | ||
| // Settled on the lane's thread, so a hold made before the UI hop still sees the failure. | ||
| failures | ||
| .Select(Settle) | ||
| .Where(onFailed => onFailed is not null) | ||
| .ObserveOn(RxSchedulers.MainThreadScheduler) | ||
| .Subscribe(onFailed => onFailed!()) | ||
| .DisposeWith(_subscriptions); | ||
| directory?.Rows.Connect() | ||
| .Subscribe(changes => { | ||
| foreach (var change in changes) | ||
| if (change.Reason is ChangeReason.Add or ChangeReason.Update | ||
| && change.Current is { IsPlaceholder: false, Origin: AgentOrigin.Local } row) | ||
| Confirm(row.Id); | ||
| }) | ||
| .DisposeWith(_subscriptions); | ||
| } | ||
|
|
||
| /// onFailed runs on the UI thread, with the worded failure, when a failure for the agent arrives | ||
| /// after this hold. | ||
| public HandoffHold Hold(string sourceSessionId, string agentId, HandoffDraft draft, Action<string>? onFailed = null) { | ||
| var id = Key(agentId); | ||
| if (RowExists(id)) return HandoffHold.AlreadyStarted; | ||
| lock (_lock) { | ||
| var now = _time.GetUtcNow(); | ||
| if (_recentFailures.Remove(id, out var failure) && now - failure.At <= RecentFailureTtl) { | ||
| _failed[sourceSessionId] = draft with { Error = HandoffFailureText.For(failure.Reason) }; | ||
| return HandoffHold.AlreadyFailed; | ||
| } | ||
| _pending[id] = new Held(sourceSessionId, draft, now, onFailed); | ||
| } | ||
| // The row can land between the first check and the registration. | ||
| if (!RowExists(id)) return HandoffHold.Pending; | ||
| Confirm(id); | ||
| return HandoffHold.AlreadyStarted; | ||
| } | ||
|
|
||
| /// The source session of a draft a later failure handed back, on the UI thread, after the | ||
| /// hold's onFailed. Whichever composer calls TakeFailed first gets the draft. | ||
| public IObservable<string> FailedFor => _failedFor; | ||
|
|
||
| /// A draft whose composer is gone before its launch settled, for the source's next composer. | ||
| public void HoldFailed(string sourceSessionId, HandoffDraft draft) { | ||
| lock (_lock) _failed[sourceSessionId] = draft; | ||
| } | ||
|
Comment on lines
+74
to
+76
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
If the user navigates away during an upload or launch and reopens the source before that operation fails, the new composer can finish Useful? React with πΒ / π. |
||
|
|
||
| public HandoffDraft? TakeFailed(string sourceSessionId) { | ||
| lock (_lock) return _failed.Remove(sourceSessionId, out var draft) ? draft : null; | ||
| } | ||
|
|
||
| public void Confirm(string agentId) { | ||
| var id = Key(agentId); | ||
| lock (_lock) { | ||
| _pending.Remove(id); | ||
| _recentFailures.Remove(id); | ||
| } | ||
| } | ||
|
|
||
| Action? Settle(LaunchFailure failure) { | ||
| var id = Key(failure.AgentId); | ||
| lock (_lock) { | ||
| var now = _time.GetUtcNow(); | ||
| if (_pending.Remove(id, out var held) && now - held.At <= PendingTtl) { | ||
| var worded = HandoffFailureText.For(failure.Reason); | ||
| _failed[held.SourceSessionId] = held.Draft with { Error = worded }; | ||
| return () => { | ||
| _directory?.RemovePlaceholder(id); | ||
| held.OnFailed?.Invoke(worded); | ||
| _failedFor.OnNext(held.SourceSessionId); | ||
| }; | ||
| } | ||
| _recentFailures[id] = (failure.Reason, now); | ||
| foreach (var stale in _recentFailures.Where(kv => now - kv.Value.At > RecentFailureTtl).Select(kv => kv.Key).ToList()) | ||
| _recentFailures.Remove(stale); | ||
| foreach (var stale in _pending.Where(kv => now - kv.Value.At > PendingTtl).Select(kv => kv.Key).ToList()) | ||
| _pending.Remove(stale); | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| bool RowExists(string id) => | ||
| _directory?.Rows.Items.Any(r => r is { IsPlaceholder: false, Origin: AgentOrigin.Local } && Key(r.Id) == id) == true; | ||
|
|
||
| static string Key(string agentId) => AgentIds.Normalize(agentId) ?? agentId; | ||
|
|
||
| public void Dispose() { | ||
| _subscriptions.Dispose(); | ||
| _failedFor.Dispose(); | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| namespace Capacitor.App.Services; | ||
|
|
||
| /// What holding a hand-off draft found. AlreadyFailed: a failure for the agent arrived before the | ||
| /// hold, and the draft waits in TakeFailed. AlreadyStarted: the agent's row is already listed. | ||
| public enum HandoffHold { Pending, AlreadyFailed, AlreadyStarted } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| using System.Reactive.Linq; | ||
| using Capacitor.Cli.Core.Handoff; | ||
| using Capacitor.Cli.Core.LocalIpc; | ||
|
|
||
| namespace Capacitor.App.Services; | ||
|
|
||
| /// Follows the daemon's stored hand-off outcome for a session id. The daemon keeps it by the | ||
| /// successor's id, so the answer is the same whether the view is the live agent or the server row | ||
| /// the successor became. | ||
| public sealed class HandoffOutcomeReader(IDaemonClientService daemon, ILocalControlOps ops, TimeProvider time) { | ||
| internal static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(2); | ||
| internal static readonly TimeSpan NullWindow = TimeSpan.FromMinutes(2); | ||
|
|
||
| /// Emits each outcome that differs in value from the last: every read deserializes fresh lists, | ||
| /// so reference equality would re-emit each poll. An answer of null (nothing stored yet, or a | ||
| /// failed read) is retried for <see cref="NullWindow"/>; a running outcome is followed without limit. | ||
| /// Completes on a terminal state or when the window lapses, and emits nothing for a daemon | ||
| /// without handoff/1. | ||
| public IObservable<HandoffOutcomeDto> Watch(string sessionId) => | ||
| daemon.Status | ||
| .SkipWhile(status => status.State != AttachState.Connected) | ||
| .Take(1) | ||
| .SelectMany(status => status.Capabilities?.Contains(HandoffWire.Capability) == true | ||
| ? Poll(sessionId) | ||
|
Comment on lines
+21
to
+24
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
When the workspace opens while the first connected daemon lacks Useful? React with πΒ / π. |
||
| : Observable.Empty<HandoffOutcomeDto>()); | ||
|
|
||
| /// The chat card's stream for one session id, inside a Switch over ids: it opens with null so a | ||
| /// card from the previous id is cleared rather than carried over. | ||
| public static IObservable<HandoffOutcomeDto?> CardFor(HandoffOutcomeReader reader, string? sessionId) => | ||
| sessionId is null | ||
| ? Observable.Return<HandoffOutcomeDto?>(null) | ||
| : reader.Watch(sessionId).Select(outcome => (HandoffOutcomeDto?)outcome).StartWith((HandoffOutcomeDto?)null); | ||
|
|
||
| IObservable<HandoffOutcomeDto> Poll(string sessionId) => | ||
| Observable.Create<HandoffOutcomeDto>(async (observer, ct) => { | ||
| var started = time.GetTimestamp(); | ||
| var seen = false; | ||
| HandoffOutcomeDto? last = null; | ||
| try { | ||
| while (!ct.IsCancellationRequested) { | ||
| var outcome = await Task.Run(() => ops.ReadHandoffOutcomeAsync(sessionId, ct), ct).ConfigureAwait(false); | ||
| if (outcome is not null) { | ||
| seen = true; | ||
| if (!Same(outcome, last)) observer.OnNext(outcome); | ||
| last = outcome; | ||
| if (outcome.State != "running") break; | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
When the daemon restarts during a takeover, Useful? React with πΒ / π. |
||
| } else if (!seen && time.GetElapsedTime(started) >= NullWindow) { | ||
| break; | ||
| } | ||
| await Task.Delay(PollInterval, time, ct).ConfigureAwait(false); | ||
| } | ||
| observer.OnCompleted(); | ||
| } catch (OperationCanceledException) when (ct.IsCancellationRequested) { | ||
| } catch (Exception ex) { | ||
| observer.OnError(ex); | ||
| } | ||
| }); | ||
|
|
||
| internal static bool Same(HandoffOutcomeDto a, HandoffOutcomeDto? b) => | ||
| b is not null && a.SourceSessionId == b.SourceSessionId && a.SourceTitle == b.SourceTitle | ||
| && a.State == b.State && a.Detail == b.Detail && a.Items.SequenceEqual(b.Items); | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If
LaunchFailedarrives beforeStartAsyncreturns but the hub invocation takes more than 30 seconds, this age check discards the buffered failure and registers the already-failed launch as pending. The caller then adds and opens a placeholder, while the source draft is never restored because that failure broadcast is not repeated. Since the launch invocation has no matching 30-second timeout, retain an early failure until itsHoldconsumes it or use a lifetime at least as long as the launch can remain in flight.Useful? React with πΒ / π.