1
>
/*---------------------------------------------------------------------------------------------
agentSubscription.ts
2
>
* Copyright (c) Microsoft Corporation. All rights reserved.
3
>
* Licensed under the MIT License. See License.txt in the project root for license information.
4
>
*--------------------------------------------------------------------------------------------*/
5
>
6
>
import { assertNever } from '../../../../base/common/assert.js';
7
>
import { Emitter, Event } from '../../../../base/common/event.js';
8
>
import { Disposable, IReference } from '../../../../base/common/lifecycle.js';
9
>
import { ResourceMap } from '../../../../base/common/map.js';
10
>
import { IObservable, observableFromEvent } from '../../../../base/common/observable.js';
11
>
import { URI } from '../../../../base/common/uri.js';
12
>
import { ActionEnvelope, ActionType, ChangesetAction, ChatAction, AnnotationsAction, ClientAnnotationsAction, ClientChangesetAction, IRootConfigChangedAction, SessionAction, StateAction, isChangesetAction, isChatAction, isAnnotationsAction, isSessionAction } from './sessionActions.js';
13
>
import { changesetReducer, chatReducer, annotationsReducer, rootReducer, sessionReducer } from './sessionReducers.js';
14
>
import { terminalReducer } from './protocol/reducers.js';
15
>
import type { RootAction, SessionAction as IProtocolSessionAction, ChatAction as IProtocolChatAction, TerminalAction } from './protocol/action-origin.generated.js';
16
>
import type { AnnotationsState, ChangesetState, ChatState, RootState, SessionState, TerminalState } from './protocol/state.js';
17
>
import type { IStateSnapshot } from './sessionProtocol.js';
18
>
import { isAhpRootChannel, ROOT_STATE_URI, StateComponents } from './sessionState.js';
19
>
20
>
// --- Public API --------------------------------------------------------------
21
>
22
>
/**
23
>
* A read-only subscription to an agent host resource (root, session, or terminal).
24
>
*
25
>
* Subscriptions are hydrated from an initial server snapshot and kept in sync
26
>
* via action envelopes. Session subscriptions support write-ahead
27
>
* reconciliation — optimistic state is layered on top of confirmed state.
28
>
*/
29
>
export interface IAgentSubscription<T> {
30
>
/**
31
>
* The current state value. For write-ahead subscriptions (sessions) this
32
>
* reflects the optimistic state (confirmed + pending replayed). For
33
>
* server-only subscriptions (root, terminal) this equals `verifiedValue`.
34
>
*
35
>
* `undefined` until the first snapshot arrives. An `Error` if subscription
36
>
* failed.
37
>
*/
38
>
readonly value: T | Error | undefined;
39
>
40
>
/**
41
>
* The server-confirmed state with no pending optimistic actions applied.
42
>
* `undefined` until the first snapshot arrives.
43
>
*/
44
>
readonly verifiedValue: T | undefined;
45
>
46
>
/** Fires when {@link value} changes (optimistic or confirmed). */
47
>
readonly onDidChange: Event<T>;
48
>
49
>
/** Fires when the subscription enters an error state. */
50
>
readonly onDidError?: Event<Error>;
51
>
52
>
/** Fires before a server-originated action is applied to this subscription's state. */
53
>
readonly onWillApplyAction: Event<ActionEnvelope>;
54
>
55
>
/** Fires after a server-originated action is applied to this subscription's state. */
56
>
readonly onDidApplyAction: Event<ActionEnvelope>;
57
>
}
58
>
59
>
/**
60
>
* Read-only snapshot describing a single active resource subscription. Used by
61
>
* inspection/debug surfaces that enumerate everything a connection is currently
62
>
* subscribed to. Does not include the always-live root state.
63
>
*/
64
>
export interface IActiveSubscriptionInfo {
65
>
/** The protocol resource URI subscribed to. */
66
>
readonly resource: URI;
67
>
/** Which state component this subscription tracks. */
68
>
readonly kind: StateComponents;
69
>
/** Number of outstanding {@link IReference} holders. */
70
>
readonly refCount: number;
71
>
/**
72
>
* The named owners currently holding a reference to this subscription,
73
>
* with how many references each holds. Names come from the `owner`
74
>
* argument passed to {@link AgentSubscriptionManager.getSubscription}.
75
>
*/
76
>
readonly holders: readonly IActiveSubscriptionHolder[];
77
>
/**
78
>
* Lifecycle status derived from the subscription's value:
79
>
* `pending` before the first snapshot, `error` if it failed, otherwise
80
>
* `snapshot`.
81
>
*/
82
>
readonly status: 'pending' | 'snapshot' | 'error';
83
>
}
84
>
85
>
/** A named owner holding one or more references to a subscription. */
86
>
export interface IActiveSubscriptionHolder {
87
>
readonly owner: string;
88
>
readonly count: number;
89
>
}
90
>
91
>
// --- Base Implementation -----------------------------------------------------
92
>
93
>
/**
94
>
* Base class for agent subscriptions. Handles envelope reception, confirmed
95
>
* state management, and action event emission.
96
>
*
97
>
* Subclasses provide the reducer and optionally override reconciliation
98
>
* behavior.
99
>
*/
100
>
abstract class BaseAgentSubscription<T> extends Disposable implements IAgentSubscription<T> {
101
>
102
>
protected _confirmedState: T | undefined;
103
>
private _error: Error | undefined;
104
>
private _bufferedEnvelopes: ActionEnvelope[] | undefined;
105
>
106
>
protected readonly _onDidChange = this._register(new Emitter<T>());
107
>
readonly onDidChange: Event<T> = this._onDidChange.event;
108
>
109
>
protected readonly _onDidError = this._register(new Emitter<Error>());
110
>
readonly onDidError: Event<Error> = this._onDidError.event;
111
>
112
>
protected readonly _onWillApplyAction = this._register(new Emitter<ActionEnvelope>());
113
>
readonly onWillApplyAction: Event<ActionEnvelope> = this._onWillApplyAction.event;
114
>
115
>
protected readonly _onDidApplyAction = this._register(new Emitter<ActionEnvelope>());
116
>
readonly onDidApplyAction: Event<ActionEnvelope> = this._onDidApplyAction.event;
117
>
118
>
protected readonly _clientId: string;
119
>
protected readonly _log: (msg: string) => void;
120
>
121
>
constructor(clientId: string, log: (msg: string) => void) {
122
super();
123
this._clientId = clientId;
124
this._log = log;
125
}
127
>
get value(): T | Error | undefined {
128
if (this._error) {
129
return this._error;