Fix native Windows PTY startup query handling (#9500)

* Fix native Windows PTY startup query handling

* Fix daemon boot smoke protocol lookup

* Fix Windows daemon repro protocol lookup
This commit is contained in:
Jinwoo Hong 2026-07-19 20:25:52 -07:00 committed by GitHub
parent d628c9300e
commit c0f0810dd9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
51 changed files with 2415 additions and 497 deletions

View File

@ -36,10 +36,11 @@ function log(message) {
// Why: the daemon rejects a hello whose protocol version differs, so read the
// current version from source rather than hardcoding a number that can drift.
function readProtocolVersion() {
const source = readFileSync(join(projectDir, 'src/main/daemon/types.ts'), 'utf8')
const protocolSourcePath = 'src/main/daemon/daemon-protocol-version.ts'
const source = readFileSync(join(projectDir, protocolSourcePath), 'utf8')
const match = source.match(/PROTOCOL_VERSION\s*=\s*(\d+)/)
if (!match) {
throw new Error('could not read PROTOCOL_VERSION from src/main/daemon/types.ts')
throw new Error(`could not read PROTOCOL_VERSION from ${protocolSourcePath}`)
}
return Number(match[1])
}

View File

@ -22,7 +22,10 @@ function log(message) {
}
function readProtocolVersion() {
const source = readFileSync(join(projectDir, 'src/main/daemon/types.ts'), 'utf8')
const source = readFileSync(
join(projectDir, 'src/main/daemon/daemon-protocol-version.ts'),
'utf8'
)
const match = source.match(/PROTOCOL_VERSION\s*=\s*(\d+)/)
if (!match) {
throw new Error('Could not read the daemon protocol version')

View File

@ -0,0 +1,523 @@
# Windows ConPTY Startup Query and Focus Authority Design
Date: 2026-07-19
Status: Final decision; implementation pending
## Problem
On native Windows ConPTY sessions, a new agent can sometimes show terminal protocol bytes as user
text:
```text
]10;rgb:2e2e/3434/3434\]11;rgb:ffff/ffff/ffff\
```
An independent symptom can prefix user input with `[I`, the printable tail of the standard terminal
focus-in report `CSI I`.
The OSC text is an Orca-generated reply to the agent's OSC 10/11 foreground/background query. The
agent can issue that query before a daemon-backed `spawn()` resolves to the renderer, and waits only
about 100 ms for the answer. Orca therefore has a short-lived main-side startup responder in
addition to the normal renderer/model query authorities. On affected ConPTY timing, the reply sent
to the PTY is returned as cooked output with its ESC bytes removed. The current ingestion order
records that cooked echo in the authoritative runtime model before any renderer-bound filtering.
The focus symptom must not be treated as the same bug without evidence. ConPTY deliberately emits
DECSET 1004 (focus reporting) and DECSET 9001 (Win32 input mode) at startup. A direct native
PowerShell ConPTY capture consistently produced the bootstrap pair, while an injected `CSI I` was
consumed as a focus event and did not render as `[I`. The bootstrap is valid transport protocol;
the failure requires an additional agent, timing, input-mode, or replay condition.
## Root Cause
The startup OSC responder currently runs in Electron main only after runtime ingestion
([`pty.ts`](../../../src/main/ipc/pty.ts)). `LocalPtyProvider` calls the runtime before its public
data listeners ([`local-pty-provider.ts`](../../../src/main/providers/local-pty-provider.ts)), while
daemon `Session` advances sequence state, writes its emulator, persists pending output, and fans out
data before Electron main receives it ([`session.ts`](../../../src/main/daemon/session.ts)). A
renderer-only filter can therefore hide the symptom without removing it from the authoritative
model, daemon history, snapshots, or remote delivery.
The corrupted echo is timing-dependent but its ordering failure is deterministic: any sanitizer
downstream of an authoritative consumer is too late. The independent `[I` symptom has not yet met
that evidentiary bar, so this design fixes the proven OSC path and adds the real focus-path harness
without pre-approving a speculative focus workaround.
## Data Flow
```text
node-pty / remote relay PTY
-> shell-ready marker scan
-> source-owned serialized ingress transaction
-> consume an early OSC query and write one canonical reply
-> match or losslessly release the measured native-Windows echo projection
-> authoritative emulator + persistence/history
-> runtime side effects and mobile/remote stream
-> renderer delivery or hidden-drop decision
-> live terminal / snapshot restore
```
Raw sequence spans travel beside cleaned strings through every downstream hop. Empty transformed
spans advance sequence and flow-control state without writing bytes into an emulator or view.
## Rejected Prototype
The current working-tree prototype is not the implementation of this design:
- It removes ConPTY's leading `?1004h` in the renderer. This changes valid native-console focus
semantics for every Windows PTY, including programs unrelated to agents.
- It filters OSC echo text only after runtime/model ingestion. The live pane can look clean while a
hidden restore, reconnect, mobile view, or CLI snapshot still contains the garbage.
- Its echo state activates only after both color slots are answered. An echo of the first response
can escape if it arrives before the second query.
- Its partial-match buffer can be lost on timeout, and its substring search can remove a later
legitimate string that happens to equal a reply.
Implementation starts by removing the prototype's bootstrap output filter and renderer-only cooked
echo filtering. Existing unrelated Windows focus-idle safeguards remain.
## Existing Contracts Preserved
This design extends, rather than replaces:
- [`terminal-model-view-contract.md`](../terminal-model-view-contract.md), especially singular query
authority, authoritative model restore, and raw sequence ordering;
- [`terminal-query-authority.md`](../terminal-query-authority.md), especially delivered-versus-
hidden-dropped ownership and replay silence;
- [`terminal-side-effect-authority.md`](../terminal-side-effect-authority.md), especially parsing PTY
bytes once in main before renderer delivery.
The startup responder is a bounded exception needed before normal delivered/dropped ownership can be
established. It must join the same ingestion decision, not operate as an unrelated renderer scrub.
## Decision 1: Source-Owned PTY Ingress Transaction
Sanitization must run on the host that owns the PTY, before that host mutates any authoritative
model or persistence. Electron main is too late for daemon sessions: `Session` has already advanced
its sequence, written its emulator, recorded pending output, and broadcast the data before main's
provider callback runs.
The transaction therefore has three installations of the same shared state machine:
- in `LocalPtyProvider`, before its configured runtime callback and data listeners;
- in daemon `Session`, before `outputSequence`, emulator writes, pending/checkpoint records, and
attached-client fanout;
- in relay `PtyHandler`, before relay replay/history buffering and `pty.data` fanout. Main's
`SshPtyProvider` is only an RPC proxy and never sanitizes relay output.
Electron main receives already-classified daemon data. A remote-runtime desktop client does not
reinterpret the stream; the remote Orca host owns its transaction. WSL sessions follow their actual
provider owner but never enable the native-Windows compatibility projection.
Fresh-session creation carries startup-transaction intent atomically. Daemon `createOrAttach`
receives the recognized-agent intent, execution-host kind, deadline, and validated renderer-pushed
color attributes. It first decides fresh versus reattach; only a fresh result constructs the
transaction before releasing the subprocess's already-buffered early output. A reattach discards
the intent without arming state. Cancellation, spawn failure, and teardown clear intent before any
PTY id can be reused.
Local creation installs the transaction before subscribing to node-pty output. Relay `pty.spawn`
carries the same fresh-session intent and installs it before releasing relay PTY output; relay
`pty.attach` never accepts it. If any owner cannot establish this ordering, the early responder is
not armed and normal query authority handles the query.
Increment `DAEMON_PROTOCOL_VERSION` for the new create/wire shape and add a named numeric-threshold
predicate in the daemon adapter/router. Current exact-version hello behavior remains; supported
legacy versions route through their existing adapters and fail the predicate. The SSH relay gets a
separate versioned spawn/data capability because daemon protocol support says nothing about relay
support.
An old daemon or relay keeps the legacy behavior and is never treated as having sanitized snapshots
or history. Main must not apply a second best-effort scrub to legacy output. New sessions receive
the invariant only after the owning daemon/relay is upgraded or restarted; an already-attached
legacy session remains explicitly outside it.
### Composition with shell-ready preprocessing
The ingress sequence domain begins after the existing shell-ready scanner. That scanner may hold
transport bytes and removes its private ready marker; marker bytes have never belonged to terminal
model or delivery sequence space and continue not to count. Its released non-marker bytes enter the
new ingress transaction in original order.
Snapshot and teardown barriers drain in pipeline order: the shell-ready scanner first releases its
non-marker buffer, then the ingress transaction resolves or abandons its candidate, then the model,
persistence, and views observe the resulting emissions. No later stage may introduce an unmetered
string transform.
### Raw sequence and emission contract
The state machine accepts source chunks with an explicit raw half-open range and returns zero or
more ordered emissions:
```ts
type PtyIngressSourceChunk = {
data: string
rawStartSeq: number
rawEndSeq: number
}
type PtyIngressEmission = {
data: string
rawStartSeq: number
rawEndSeq: number
transformed: boolean
}
```
The raw range is the post-shell-ready ingress sequence domain and remains contiguous even when
`data` is shorter after sanitization. The ordered emissions partition accepted source ranges without
overlap. Producer flow-control acknowledgement follows those emitted raw spans and is counted once
from `rawEndSeq - rawStartSeq`, never from emitted string length. A held prefix delays its ACK; the
buffer is strictly bounded and the serialized queue cannot acknowledge later output ahead of it.
The ingress raw high-water advances when a source chunk is accepted. Model-applied and
view-delivered high-waters advance only through ordered emissions, including empty transformed
emissions that consume a raw range. Runtime/mobile listener metadata carries
`rawLength = rawEndSeq - rawStartSeq`, not `data.length`. A restore that overlaps a transformed span
cannot slice cleaned text by raw offset and must request a fresh authoritative snapshot, matching
the existing `rawLength !== data.length` safety rule.
This metadata is end-to-end, not local to the state machine. Implementation changes the complete
path:
1. daemon `Session` or relay `PtyHandler` emission callback;
2. daemon/relay batching, coalescing, splitting, and wire notification;
3. daemon adapter or `SshPtyProvider` decode;
4. `OrcaRuntimeService` model and runtime/mobile listeners;
5. main renderer batching, pending/drop accounting, and preload payload;
6. renderer reconciliation and remote-runtime binary/live frame decoding.
The wire representation carries `data`, `seq = rawEndSeq`, `rawLength`, and `transformed`. A
span-only emission has empty `data` but non-zero `rawLength`; no layer may drop it before advancing
its high-water and producer ACK. It is not written to an emulator or xterm. Coalescing is permitted
only for contiguous spans and sums raw lengths independently of string lengths. A transformed
emission is indivisible because there is no byte-for-byte raw-to-clean offset; splitters must flush
it as its own frame or request snapshot reconciliation instead of slicing it.
If a snapshot is requested while a partial candidate is held, the transaction first abandons that
candidate and releases its bytes unchanged as an ordered emission. The authoritative emulator and
snapshot sequence therefore describe the same raw high-water.
### Serialization and teardown
Each PTY has a non-reentrant serialized ingress queue. A provider write may synchronously produce a
nested callback, but that callback is appended after the current source chunk rather than delivered
ahead of its remaining bytes. Timeout releases and snapshot barriers enter the same queue.
On exit, the queue releases all buffered bytes, applies those emissions to the authoritative model
and persistence, and fans them out before `onPtyExit`, `pty:exit`, or PTY state cleanup. Relay
disposal must flush both ingress prefixes and its existing pending output batches before clearing
them or killing PTYs, matching the natural-exit flush-before-`pty.exit` order. No drain may recreate
state after teardown. Ordinary chunks pass through as one unchanged emission.
## Decision 2: Startup OSC Queries Are Consumed Authoritatively
The early OSC 10/11 responder remains because removing it would regress daemon-hosted agent startup
and the agent's short color-query timeout. Its implementation moves into the source-owned ingress
transaction.
When registered for an agent spawn, it:
1. recognizes exact OSC 10/11 query grammar across provider chunks;
2. builds replies from validated renderer-pushed foreground/background attributes;
3. emits the canonical ST-terminated, 16-bit-channel reply used by renderer/model query authority,
regardless of whether the query ended with BEL or ST;
4. records each reply transaction before writing it to the provider;
5. consumes the answered query from the authoritative model and view emissions so neither the
hidden model nor a delivered renderer can answer it a second time;
6. begins echo recognition as soon as each individual reply is written.
If attributes or a provider are unavailable, the query is not consumed. It passes through unchanged
to normal query authority. A reattach never registers startup response state, matching current
behavior.
### Exact authority transfer
Startup query interception opens only for a fresh session whose atomic creation installed the
transaction before buffered output release. It closes at the first of:
- both OSC 10 and OSC 11 slots have been answered;
- the startup deadline expires;
- main sends an ordered authority-close control after either the consuming-view handshake or the
hidden-runtime ownership mark is established;
- the spawn fails, is cancelled, reattaches, or exits.
Closing query interception does not discard already-written reply candidates. Echo recognition has
its own bounded lifetime and may finish or drain after normal authority takes over. A close and a
provider callback are ordered by the source owner's per-PTY ingress queue. Transport attachment to
a daemon/relay client is not a consuming-view signal. Main sends the close over a versioned control
method, and the source owner acknowledges the applied ingress sequence. Queries before that ordered
boundary are either consumed at source or removed from emissions; queries after it pass unchanged
to the normal delivered/hidden decision. Each query therefore belongs to exactly one authority.
The regular authority rules continue after the bounded startup window:
- delivered live bytes are answered by the live view;
- hidden-dropped live bytes are answered by the runtime model;
- replayed, seeded, and snapshot bytes are answered by nobody;
- the daemon's persistence emulator never writes replies.
Implementation must amend `terminal-model-view-contract.md` and `terminal-query-authority.md` to
name this source-owner startup authority, its opening/closing events, and its no-replay rule. It is a
real third responder class, not an undocumented exception.
## Decision 3: Matched Echo Suppression Is Lossless and Pre-Model
Only native Windows ConPTY agent spawns that pass the deterministic provider harness enable a
compatibility projection for replies written by the startup transaction. WSL and POSIX SSH PTYs do
not. A remote-runtime PTY can enable it only on its owning Windows host under the same evidence and
capability gate.
The harness records the exact projection ConPTY returns for the canonical ST reply, including its
chunking and any console transformation. The implementation must not assume that the projection is
always merely "remove ESC", and BEL or other reply forms are not added without separate evidence.
For every written reply, the transaction records that exact expected projection. Recognition is:
- FIFO in reply-write order;
- anchored at the next possible output position, not an unbounded substring search;
- active immediately for each reply instead of waiting for both color slots;
- bounded by the startup deadline and maximum reply length;
- streaming across chunk boundaries.
If incoming bytes diverge from the expected projection, all buffered bytes are released unchanged
and that candidate is abandoned. If the deadline expires or the PTY is cleared while a prefix is
buffered, the prefix is released through the serialized ingestion queue; it is never discarded. A
match advances the raw span but emits no bytes to the model, persistence, or views.
Projection matching cannot prove provenance. An application can print the same projected text at
the candidate position, causing a false-positive removal while the later real echo remains visible.
This is the central drawback of the workaround. It is accepted only if the real provider harness
shows a stable, immediate projection inside the narrow registered-agent startup window; otherwise
Orca disables the projection and keeps the visible output rather than risking deletion.
When the projection is enabled, the exact-collision behavior is explicit: the first identical
anchored candidate is removed and a later real echo is allowed through. The test fixture must assert
that result. This accepts a narrowly bounded false-positive risk instead of pretending provenance
is knowable; the release gate must document the observed timing window and justify that tradeoff.
Unknown, delayed, or interleaved transformations pass through visibly.
## Decision 4: Preserve ConPTY Focus Protocol
Orca must deliver the ConPTY bootstrap `?1004h`/`?9001h` to live terminal emulators unchanged. It
must not remove, reorder, or fabricate transport bootstrap modes.
The `[I` investigation gets a deterministic harness before a behavior change. The harness must
exercise the actual renderer focus callback and provider write path, not only inject `CSI I`
directly into node-pty. It records:
- raw provider output and its order;
- whether the focus report came from live ConPTY bootstrap state, an application-owned DECSET 1004,
or replayed snapshot state;
- the exact bytes written to the provider;
- the exact bytes returned by the provider and stored in the model;
- agent lifecycle state when the report was emitted.
The implementation gate is strict:
- If stale snapshot modes cause the report, fix snapshot rehydration. Transport bootstrap focus mode
is not persisted as application ownership.
- If a live agent receives transport focus before it owns terminal focus reporting, record that
evidence without shipping a suppression heuristic from this document.
- If provider input/output transformation corrupts a correctly owned focus report, fix or sanitize
that transformation at the same pre-model ingress boundary used for OSC replies.
No global `?1004h` filter ships under any outcome.
If the harness proves a focus-ownership race, a follow-up design is required before implementation.
That design must define the output-to-input ownership signal, distinguish xterm focus events from
identical typed/pasted/programmatic bytes, specify startup transitions and deadlines, define any
snapshot/wire metadata, and cover the separate explicit reattach-focus write path. This document
does not pre-approve an ownership state machine whose input provenance cannot yet be represented.
## Snapshot and Replay Rules
Interactive modes in a snapshot are capabilities of the live application, not proof that a new
view should emit input immediately.
- Cold restore into a fresh shell keeps the existing full mode reset.
- Reattach to a live agent may rehydrate application-owned focus mode, but never transport-only
bootstrap ownership.
- Snapshot serialization/replay must not mutate live ownership trackers.
- A snapshot containing an OSC query or an old Orca reply never produces a provider write.
- Model and renderer snapshots must both be free of matched cooked reply projections.
Any future focus ownership metadata must be explicit; it cannot be inferred from serialized
`?1004h` text.
## Failure Policy
Safety is asymmetric:
- This design does not intentionally suppress a focus notification; evidence of a focus-ownership
race triggers a follow-up design instead.
- Passing through an unrecognized OSC echo is safer than deleting output that may belong to the
application.
- Missing a startup color reply falls back to the existing renderer/model authority. Duplicate
replies are forbidden.
- Losing buffered output on timeout or teardown is forbidden.
## Cross-Platform and Remote Scope
- Native Windows local and daemon ConPTY: startup response plus evidence-gated echo compatibility
path; focus investigation only until a proven root cause has its own complete design.
- WSL: normal Linux terminal semantics; no ConPTY echo or focus workaround.
- SSH: the same ingress/query ownership ordering, but no native Windows echo projection unless the
remote host protocol later supplies explicit equivalent evidence.
- Remote runtime: the remote Orca host owns ingestion and must implement the same contract there;
desktop local main does not reinterpret its stream.
- Mobile/web views: consume the sanitized authoritative model stream and retain exactly-one query
response authority through the existing terminal-driver election.
## Edge Cases
- A query or echo split at any byte boundary, including OSC ST split across chunks.
- The first reply echo arriving before the second color query.
- Unrelated output, an exact application-text collision, or a partial match before the real echo.
- Timeout, snapshot, detach, process exit, daemon shutdown, or relay disposal while bytes are held.
- A provider write causing a synchronous nested output callback.
- Fresh spawn versus reattach, cancellation, PTY-id reuse, and an authority-close control racing
with output.
- BEL-terminated queries still receiving the canonical ST reply.
- Empty transformed spans crossing coalescing, splitting, ACK, mobile, and remote-runtime layers.
- Old daemon/relay protocol versions and sessions that survive an application upgrade.
- WSL or POSIX SSH running from a Windows desktop without inheriting local ConPTY workarounds.
- Focus gained through normal terminal input versus the separate explicit reattach-focus write path.
## Test Plan
### Deterministic provider harness
- Capture the native ConPTY bootstrap across natural and forced chunk boundaries.
- Reproduce the OSC reply echo with the real provider and an agent query fixture.
- Exercise BEL and ST queries and assert the same canonical ST reply, plus separate OSC 10/11
queries and combined OSC 10 `?;?`.
- Force the first reply echo before the second query.
- Split every byte boundary in query and echo fixtures.
- Print the expected projection immediately before the real echo and assert that the first
identical candidate is removed while the later echo passes through.
- Exercise focus gain/loss through the actual xterm focus path.
### Ingress integration
- Assert local runtime ingestion and daemon emulator/pending/checkpoint persistence receive
classified data before storing it.
- Assert renderer and mobile delivery receive the same visible output.
- Assert raw start/end spans advance by original provider length, producer ACK contribution is
counted exactly once, and mobile/runtime metadata reports the raw span rather than string length.
- Assert a partial candidate is released unchanged on mismatch, timeout, snapshot barrier, move,
and teardown, including a prefix held from an earlier callback.
- Assert re-entrant provider callbacks remain ordered behind the source chunk that caused the
write.
- Assert restore overlap across a transformed or delayed span requests a fresh snapshot rather
than slicing cleaned text.
- Assert `LocalPtyProvider`, daemon `Session`, and relay `PtyHandler` install the source-side seam;
assert `SshPtyProvider` does not reinterpret relay data.
- Assert WSL, SSH, and remote-runtime streams do not enable the native-Windows projection.
- Assert a capable daemon sanitizes live output, snapshots, pending records, checkpoints, and cold
restore; assert capable relay replay/history has the same property. Assert old daemon and relay
versions are detected and never represented as sanitized.
### Authority and restore
- Assert an early-consumed query is answered once by its source owner and never by renderer or
hidden model.
- Assert normal delivered/dropped query authority resumes after startup state clears.
- Assert the main snapshot, renderer snapshot, hidden reveal, reconnect, and mobile subscription do
not contain cooked OSC text.
- Assert replay never sends OSC or focus replies.
- Assert ordinary native console focus behavior remains enabled.
- Assert the focus harness captures both terminal `onData` and explicit reattach-focus writes. Any
later ownership implementation defines its tests in the required follow-up design.
### End-to-end acceptance
On native Windows, repeatedly create, hide, reveal, and reconnect new and resumed agent sessions.
Before typing, neither live output nor any restore path may contain `]10;rgb`/`]11;rgb`. Repeat the
focus/blur scenarios to classify `[I`; if it reproduces through an ownership race, this work stops
at the follow-up-design gate rather than claiming the symptom fixed. Application-owned focus
behavior must still move the TUI caret correctly after focus and reattach. Repeat with a plain
PowerShell terminal and a native focus-event consumer to prove no global regression.
Run renderer and main PTY suites, runtime snapshot/query suites, node/web typechecks, lint,
formatting, max-lines ratchet, reliability gates, and Electron validation. SSH ingestion changes also
require the repository's SSH end-to-end procedure.
## UI Quality Bar
This is not a layout, styling, or copy change. Existing terminal rendering and focus behavior must
look unchanged except that matched protocol garbage is absent. A passing terminal screenshot has a
clean prompt, no clipped or duplicated startup output, no restore flash, and the existing cursor and
focus presentation.
## Review Screenshots
1. A fresh native-Windows agent prompt after startup, with no OSC reply text.
2. The same session after hide/reveal restore, still clean and without duplicated output.
3. The session after focus, blur, and reconnect, showing the focus-harness outcome and prompt state.
4. A plain native PowerShell terminal after focus/blur, showing unchanged adjacent behavior.
## Rollout
1. **Correct seam and rollback.** Remove the prototype filters. Add the shared serialized ingress
state machine, explicit raw-span emissions, and separate ACK accounting after shell-ready
preprocessing. Install pass-through mode in `LocalPtyProvider`, daemon `Session`, and relay
`PtyHandler` before their models, persistence/replay, and fanout.
2. **Protocol and authority contract.** Bump the daemon protocol, add relay capability/version gates,
carry atomic fresh-spawn intent, define legacy fallback, and amend the canonical model/query
authority documents with exact transfer events.
3. **Startup transaction.** Move OSC startup recognition/reply into source ingress, retain canonical
ST replies, consume answered queries for model and views, and retain raw sequence accounting.
4. **Windows compatibility projection.** Gate the measured projection on provider evidence; add FIFO
anchored recognition, serialized re-entrant delivery, lossless drains, false-positive coverage,
and authoritative daemon/model/snapshot tests.
5. **Focus evidence.** Land the real focus-path harness. If it proves a focus-ownership race, stop
for the required follow-up design; a direct snapshot or provider-transformation bug may be fixed
only with a failing regression test that selects that branch.
6. **Electron and SSH gates.** Validate visible, hidden, restored, mobile-owned, native console, WSL,
and SSH scenarios before removing the old startup implementation.
Each slice must keep query authority singular. The compatibility projection does not ship without a
model-snapshot assertion, and no focus behavior change ships before the deterministic focus harness
fails on the old behavior and passes on the new behavior.
## Lightweight Eng Review
- Scope: limited to the proven OSC startup corruption plus deterministic focus evidence. Global
focus filtering and an unproven ownership state machine remain out of scope.
- Architecture/data flow: classification belongs at each PTY source owner after shell-ready
preprocessing and before every authoritative model, persistence, replay, or delivery consumer.
- Failure modes covered: partial/mismatched projections, false-positive collision, nested writes,
authority races, snapshot and teardown drains, protocol-version skew, reattach, and host
isolation.
- Test coverage required: byte-boundary unit tests for the shared transaction; local, daemon, relay,
runtime/mobile, restore, and legacy-protocol integration tests; real native-Windows provider and
renderer-focus harnesses; Electron and SSH end-to-end validation.
- Performance/blast radius: ordinary output is a pass-through emission. Buffering is bounded by one
startup reply candidate and its deadline. Protocol and sequence metadata touch every delivery
path, so existing high-throughput, ACK, hidden-drop, and reconnect tests are mandatory.
- UI quality bar: terminal layout and styling are unchanged; only matched startup garbage
disappears, without cursor/focus regressions or restore flashes.
- Required review screenshots: the four terminal states in `Review Screenshots`.
- Residual risks: the native projection may be too unstable to enable; exact projected application
output can collide; `[I` may require a separately reviewed focus-ownership design.
## Non-Goals
- Replacing xterm's general query parser.
- Filtering arbitrary escape-looking terminal output.
- Disabling ConPTY focus or Win32 input mode globally.
- Changing WSL, SSH, or remote-runtime terminal semantics to imitate local Windows.
- Solving unrelated MCP startup warnings reported beside the terminal garbage.
## Final Invariants
1. Provider bytes are classified once before model ingestion and view delivery.
2. A live query has exactly one responder; replay has none.
3. Source-owner persistence, runtime/model state, and every view agree on removal of a matched
startup-reply projection.
4. Raw provider sequence accounting survives sanitization.
5. Buffered unmatched output is always released; the sanitizer cannot silently lose user data.
6. ConPTY bootstrap modes reach live emulators unchanged.
7. No focus suppression ships from this design; a proven ownership race requires a follow-up design.
8. PTY teardown clears all startup-transaction buffers and focus-harness measurement state.

View File

@ -0,0 +1,12 @@
// Why: daemons survive app updates, so wire behavior must be version-gated.
export const PROTOCOL_VERSION = 24
export const PTY_STARTUP_INGRESS_PROTOCOL_VERSION = 24
export const GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION = 22
export const CLEAN_DISCONNECT_PROTOCOL_VERSION = 24
export const PREVIOUS_DAEMON_PROTOCOL_VERSIONS = [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23
] as const
export function supportsPtyStartupIngress(protocolVersion: number): boolean {
return protocolVersion >= PTY_STARTUP_INGRESS_PROTOCOL_VERSION
}

View File

@ -144,6 +144,72 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
expect(result.providerSequence).toEqual({ value: 0, generation: 'reset' })
})
it('carries classified startup spans from the daemon source to the adapter', async () => {
const onData = vi.fn()
adapter.onData(onData)
const { id } = await adapter.spawn({
cols: 80,
rows: 24,
startupIngress: {
colors: { foreground: '#2e3434', background: '#ffffff' },
deadlineMs: 5_000,
...(process.platform === 'win32'
? { echoProjection: 'windows-conpty-esc-stripped' as const }
: {})
}
})
const query = '\x1b]10;?\x07'
lastSubprocess._simulateData(query)
lastSubprocess._simulateData('prompt')
await waitFor(() => onData.mock.calls.length >= 2)
expect(lastSubprocess.write).toHaveBeenCalledWith('\x1b]10;rgb:2e2e/3434/3434\x1b\\')
expect(onData).toHaveBeenCalledWith({
id,
data: '',
sequenceChars: query.length,
seq: query.length,
transformed: true
})
expect(onData).toHaveBeenCalledWith({ id, data: 'prompt' })
await expect(adapter.getBufferSnapshot(id)).resolves.toMatchObject({
data: expect.not.stringContaining(']10;rgb')
})
})
it('omits startup intent and close control for the preserved v23 protocol', async () => {
const ensureConnectedSpy = vi
.spyOn(DaemonClient.prototype, 'ensureConnected')
.mockResolvedValue()
const requestSpy = vi.spyOn(DaemonClient.prototype, 'request').mockResolvedValue({
isNew: true,
pid: null,
shellState: 'unsupported',
snapshot: null
} as never)
const legacy = new DaemonPtyAdapter({ socketPath, tokenPath, protocolVersion: 23 })
try {
await legacy.spawn({
sessionId: 'legacy-session',
cols: 80,
rows: 24,
startupIngress: {
colors: { foreground: '#2e3434', background: '#ffffff' },
deadlineMs: 5_000
}
})
const createPayload = requestSpy.mock.calls.find(([type]) => type === 'createOrAttach')?.[1]
expect(createPayload).not.toHaveProperty('startupIngress')
await expect(legacy.closeStartupQueryAuthority('legacy-session')).resolves.toBe(0)
expect(requestSpy).not.toHaveBeenCalledWith('closeStartupQueryAuthority', expect.anything())
} finally {
legacy.dispose()
requestSpy.mockRestore()
ensureConnectedSpy.mockRestore()
}
})
it('uses worktreeId as session prefix when provided', async () => {
const result = await adapter.spawn({ cols: 80, rows: 24, worktreeId: 'wt-1' })
expect(result.id).toContain('wt-1')

View File

@ -14,6 +14,7 @@ import {
CLEAN_DISCONNECT_PROTOCOL_VERSION,
GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION,
PROTOCOL_VERSION,
supportsPtyStartupIngress,
type CreateOrAttachResult,
type DaemonEvent,
type GetSnapshotResult,
@ -105,6 +106,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}) => void)[] = []
private exitListeners: ((payload: { id: string; code: number }) => void)[] = []
private backgroundStreamListeners: ((payload: PtyBackgroundStreamEvent) => void)[] = []
@ -142,6 +145,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
// must never see them, so gating makes them silent no-ops there.
private supportsProducerFlowControl: boolean
private supportsAuthoritativeBufferSnapshots: boolean
private supportsStartupIngress: boolean
private pausedProducerSessionIds = new Set<string>()
// Why tracked here: the daemon's background set (keep-tail stream thinning
// + transient-fact scan authority) dies with the daemon process/socket;
@ -185,6 +189,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.supportsIncrementalCheckpoints = this.protocolVersion >= 13
this.supportsProducerFlowControl = this.protocolVersion >= 19
this.supportsAuthoritativeBufferSnapshots = this.protocolVersion >= 20
this.supportsStartupIngress = supportsPtyStartupIngress(this.protocolVersion)
this.client.onDisconnected(() => {
for (const id of this.pausedProducerSessionIds) {
this.producerResumesOwedOnReconnect.add(id)
@ -299,7 +304,10 @@ export class DaemonPtyAdapter implements IPtyProvider {
terminalWindowsPowerShellImplementation: opts.terminalWindowsPowerShellImplementation,
shellReadySupported,
...(shellReadyTimeoutMs !== undefined ? { shellReadyTimeoutMs } : {}),
...(historySeed ? { historySeed } : {})
...(historySeed ? { historySeed } : {}),
...(this.supportsStartupIngress && opts.startupIngress
? { startupIngress: opts.startupIngress }
: {})
})
let scrollback = restoreInfo ? getRecoveredHistorySeed(restoreInfo) : null
@ -925,7 +933,13 @@ export class DaemonPtyAdapter implements IPtyProvider {
}
onData(
callback: (payload: { id: string; data: string; sequenceChars?: number }) => void
callback: (payload: {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}) => void
): () => void {
this.dataListeners.push(callback)
return () => {
@ -1455,9 +1469,11 @@ export class DaemonPtyAdapter implements IPtyProvider {
listener({
id: event.sessionId,
data: event.payload.data,
...(event.payload.sequenceChars === undefined
...((event.payload.rawLength ?? event.payload.sequenceChars) === undefined
? {}
: { sequenceChars: event.payload.sequenceChars })
: { sequenceChars: event.payload.rawLength ?? event.payload.sequenceChars }),
...(event.payload.transformed ? { transformed: true } : {}),
...(event.payload.seq === undefined ? {} : { seq: event.payload.seq })
})
}
} else if (event.event === 'sessionBackgroundMarker') {
@ -1518,6 +1534,16 @@ export class DaemonPtyAdapter implements IPtyProvider {
}
})
}
async closeStartupQueryAuthority(id: string): Promise<number> {
if (!this.supportsStartupIngress) {
return 0
}
const result = await this.client.request<{ appliedSeq: number }>('closeStartupQueryAuthority', {
sessionId: id
})
return result.appliedSeq
}
}
// Why: ENOENT/ECONNREFUSED with syscall 'connect' mean the socket is

View File

@ -17,6 +17,8 @@ export class DaemonPtyRouter implements IPtyProvider {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}) => void)[] = []
private exitListeners: ((payload: { id: string; code: number }) => void)[] = []
@ -143,6 +145,10 @@ export class DaemonPtyRouter implements IPtyProvider {
await this.adapterFor(id).clearBuffer(id)
}
async closeStartupQueryAuthority(id: string): Promise<number> {
return (await this.adapterFor(id).closeStartupQueryAuthority?.(id)) ?? 0
}
acknowledgeDataEvent(id: string, charCount: number): void {
this.adapterFor(id).acknowledgeDataEvent(id, charCount)
}
@ -183,7 +189,13 @@ export class DaemonPtyRouter implements IPtyProvider {
}
onData(
callback: (payload: { id: string; data: string; sequenceChars?: number }) => void
callback: (payload: {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}) => void
): () => void {
this.dataListeners.push(callback)
return () => {

View File

@ -23,6 +23,8 @@ import type { SubprocessHandle } from './session'
import { checkPtySpawnHealth } from './pty-subprocess'
import { createNoopDaemonFileLog, type DaemonFileLog } from './daemon-file-log'
import { isTuiAgent } from '../../shared/tui-agent-config'
import { parsePtyStartupIngressIntent } from '../../shared/pty-startup-ingress'
import { isNativeWindowsLocalPtySpawn } from '../runtime/terminal-model-query-authority'
import { unlinkOwnedDaemonPidFile, unlinkOwnedDaemonTokenFile } from './daemon-spawner'
import {
CLEAN_DISCONNECT_PROTOCOL_VERSION,
@ -707,11 +709,18 @@ export class DaemonServer {
terminalWindowsPowerShellImplementation: p.terminalWindowsPowerShellImplementation,
shellReadySupported: p.shellReadySupported,
historySeed: p.historySeed,
startupIngress: parsePtyStartupIngressIntent(p.startupIngress, {
allowWindowsEchoProjection: isNativeWindowsLocalPtySpawn({
connectionId: null,
cwd: p.cwd,
shellOverride: p.shellOverride
})
}),
...(p.shellReadyTimeoutMs !== undefined
? { shellReadyTimeoutMs: p.shellReadyTimeoutMs }
: {}),
streamClient: {
onData: (data) => {
onData: (data, rawLength = data.length, transformed = false, seq) => {
// Scan BEFORE enqueue: the batcher may keep-tail drop this
// chunk, but its facts must be captured regardless.
this.transientFactRelay.onSessionData(p.sessionId, data)
@ -722,7 +731,10 @@ export class DaemonServer {
performance.now() - lastInputAt <= DaemonServer.INTERACTIVE_OUTPUT_WINDOW_MS
this.streamDataBatcher.enqueue(clientId, p.sessionId, data, {
flushImmediately: isInteractiveOutput,
flushMaxChars: DaemonServer.INTERACTIVE_OUTPUT_MAX_CHARS
flushMaxChars: DaemonServer.INTERACTIVE_OUTPUT_MAX_CHARS,
rawLength,
transformed,
seq
})
},
onExit: (code) => {

View File

@ -77,6 +77,23 @@ describe('DaemonStreamDataBatcher', () => {
}
})
it('preserves transformed span metadata through an immediate flush', () => {
const { batcher, streamSocket } = createBatcher()
batcher.enqueue('client-1', 'session-1', '', {
flushImmediately: true,
rawLength: 9,
seq: 17,
transformed: true
})
expect(streamSocket.write).toHaveBeenCalledTimes(1)
expect(JSON.parse(String(streamSocket.write.mock.calls[0]?.[0]))).toMatchObject({
event: 'data',
payload: { data: '', rawLength: 9, seq: 17, sequenceChars: 9, transformed: true }
})
})
it('keeps large pending output batched even when an interactive redraw follows', () => {
vi.useFakeTimers()
try {

View File

@ -13,6 +13,7 @@ import {
type PendingStreamDataBatch
} from './daemon-stream-keep-tail-drop'
import type { DaemonEvent } from './types'
import { appendDaemonStreamData, type DaemonStreamEnqueueOptions } from './daemon-stream-data-entry'
type StreamDataClient = {
streamSocket: Socket | null
@ -59,11 +60,6 @@ const HELD_WRITE_THROUGH_TOTAL_CHARS = 32 * 1024 * 1024
// sessions × this ≈ tens of KB.
const SMALL_SESSION_HOLD_BYPASS_CHARS = 4 * 1024
type EnqueueOptions = {
flushImmediately?: boolean
flushMaxChars?: number
}
type DaemonStreamDataBatcherOptions = {
maxLineBytes?: number
/** Fires after each stream-socket write the only place backlog grows, so
@ -97,26 +93,19 @@ export class DaemonStreamDataBatcher {
this.salvageDroppedData = options.salvageDroppedData ?? (() => '')
}
enqueue(clientId: string, sessionId: string, data: string, options: EnqueueOptions = {}): void {
enqueue(
clientId: string,
sessionId: string,
data: string,
options: DaemonStreamEnqueueOptions = {}
): void {
const client = this.getClient(clientId)
if (!client?.streamSocket || client.streamSocket.destroyed) {
return
}
const batch = this.getOrCreateBatch(clientId)
const last = batch.queue.at(-1)
// Never coalesce across a control entry — it marks a position in the
// session's byte order.
if (last?.sessionId === sessionId && !last.control) {
last.data += data
} else {
batch.queue.push({ sessionId, data })
}
batch.queuedChars += data.length
batch.queuedCharsBySession.set(
sessionId,
(batch.queuedCharsBySession.get(sessionId) ?? 0) + data.length
)
appendDaemonStreamData(batch, sessionId, data, options)
if (this.isSessionDroppable(sessionId)) {
// Keep-tail scales down as more backgrounded sessions queue, bounding
@ -256,12 +245,16 @@ export class DaemonStreamDataBatcher {
})
}
const end =
entry.data.length <= BULK_WRITE_SLICE_CHARS
entry.transformed || entry.data.length <= BULK_WRITE_SLICE_CHARS
? entry.data.length
: clampToSafeSplitIndex(entry.data, 0, BULK_WRITE_SLICE_CHARS)
const slice = entry.data.slice(0, end)
const entrySequenceChars = entry.sequenceChars ?? entry.data.length
const sliceSequenceChars = entrySequenceChars === 0 ? 0 : slice.length
const sliceSequenceChars = entry.transformed
? entrySequenceChars
: entrySequenceChars === 0
? 0
: slice.length
if (end >= entry.data.length) {
batch.queue.shift()
} else {
@ -278,7 +271,15 @@ export class DaemonStreamDataBatcher {
} else {
batch.queuedCharsBySession.set(entry.sessionId, sessionHeldAfter)
}
writeStreamDataEvents(socket, entry.sessionId, slice, this.maxLineBytes, sliceSequenceChars)
writeStreamDataEvents(
socket,
entry.sessionId,
slice,
this.maxLineBytes,
sliceSequenceChars,
entry.seq,
entry.transformed
)
this.onAfterSocketWrite?.()
}
if (retained.length > 0) {
@ -369,7 +370,9 @@ export class DaemonStreamDataBatcher {
entry.sessionId,
entry.data,
this.maxLineBytes,
entry.sequenceChars ?? entry.data.length
entry.sequenceChars ?? entry.data.length,
entry.seq,
entry.transformed
)
this.onAfterSocketWrite?.()
}

View File

@ -0,0 +1,46 @@
import type { PendingStreamDataBatch } from './daemon-stream-keep-tail-drop'
export type DaemonStreamEnqueueOptions = {
flushImmediately?: boolean
flushMaxChars?: number
rawLength?: number
transformed?: boolean
seq?: number
}
export function appendDaemonStreamData(
batch: PendingStreamDataBatch,
sessionId: string,
data: string,
options: DaemonStreamEnqueueOptions
): void {
const last = batch.queue.at(-1)
// Why: control and transformed spans mark indivisible source-stream positions.
if (
last?.sessionId === sessionId &&
!last.control &&
!last.transformed &&
options.transformed !== true
) {
last.data += data
const rawLengthBefore = last.sequenceChars ?? last.data.length - data.length
const combinedRawLength = rawLengthBefore + (options.rawLength ?? data.length)
last.sequenceChars = combinedRawLength === last.data.length ? undefined : combinedRawLength
last.seq = options.seq
} else {
batch.queue.push({
sessionId,
data,
...(options.rawLength === undefined || options.rawLength === data.length
? {}
: { sequenceChars: options.rawLength }),
...(options.transformed ? { transformed: true } : {}),
...(options.seq === undefined ? {} : { seq: options.seq })
})
}
batch.queuedChars += data.length
batch.queuedCharsBySession.set(
sessionId,
(batch.queuedCharsBySession.get(sessionId) ?? 0) + data.length
)
}

View File

@ -9,18 +9,26 @@ import type { Socket } from 'node:net'
export function encodeStreamDataEvent(
sessionId: string,
data: string,
sequenceChars?: number
rawLength?: number,
seq?: number,
transformed?: boolean
): string {
return encodeNdjson({
type: 'event',
event: 'data',
sessionId,
payload: { data, ...(sequenceChars === undefined ? {} : { sequenceChars }) }
payload: {
data,
...(seq === undefined ? {} : { seq }),
...(rawLength === undefined ? {} : { rawLength }),
...(rawLength === undefined ? {} : { sequenceChars: rawLength }),
...(transformed ? { transformed: true } : {})
}
})
}
function streamDataEventLineBytes(sessionId: string, data: string, sequenceChars?: number): number {
return Buffer.byteLength(encodeStreamDataEvent(sessionId, data, sequenceChars), 'utf8')
function streamDataEventLineBytes(sessionId: string, data: string, rawLength?: number): number {
return Buffer.byteLength(encodeStreamDataEvent(sessionId, data, rawLength), 'utf8')
}
function isHighSurrogate(value: number): boolean {
@ -100,15 +108,27 @@ export function writeStreamDataEvents(
sessionId: string,
data: string,
maxLineBytes: number,
sequenceChars = data.length
rawLength = data.length,
seq?: number,
transformed = false
): void {
const explicitSequenceChars = sequenceChars === data.length ? undefined : sequenceChars
for (const chunk of splitStreamDataForNdjson(
const explicitRawLength = rawLength === data.length ? undefined : rawLength
if (transformed) {
streamSocket.write(encodeStreamDataEvent(sessionId, data, rawLength, seq, true))
return
}
const carriesMetadata = explicitRawLength !== undefined || seq !== undefined
const chunks = splitStreamDataForNdjson(
sessionId,
data,
maxLineBytes,
explicitSequenceChars
)) {
streamSocket.write(encodeStreamDataEvent(sessionId, chunk, explicitSequenceChars))
carriesMetadata ? Math.max(1, maxLineBytes - 96) : maxLineBytes,
explicitRawLength
)
let consumed = 0
for (const chunk of chunks) {
consumed += chunk.length
const chunkEndSeq = seq === undefined ? undefined : seq - (data.length - consumed)
const chunkRawLength = explicitRawLength === 0 ? 0 : carriesMetadata ? chunk.length : undefined
streamSocket.write(encodeStreamDataEvent(sessionId, chunk, chunkRawLength, chunkEndSeq))
}
}

View File

@ -5,7 +5,14 @@ export type DataEvent = {
type: 'event'
event: 'data'
sessionId: string
payload: { data: string; sequenceChars?: number }
payload: {
data: string
seq?: number
rawLength?: number
transformed?: boolean
/** Legacy v23 name retained for old adapter fixtures. */
sequenceChars?: number
}
}
export type ExitEvent = {

View File

@ -21,6 +21,8 @@ export type StreamQueueEntry = {
/** Original PTY characters represented by data. Salvaged query copies are
* delivered bytes but represent zero new positions in the source stream. */
sequenceChars?: number
seq?: number
transformed?: boolean
control?: DaemonEvent
}

View File

@ -3,6 +3,7 @@ import { shutdownDegradedFallbackSessions } from './degraded-daemon-fallback-shu
import type {
IPtyProvider,
PtyBackgroundStreamEvent,
PtyDataEvent,
PtyProviderBufferSnapshot,
PtyProcessInfo,
PtySpawnOptions,
@ -26,11 +27,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
private fallback: ManagedPtyProvider
private sessionProviders = new Map<string, ManagedPtyProvider>()
private unsubscribers: (() => void)[] = []
private dataListeners: ((payload: {
id: string
data: string
sequenceChars?: number
}) => void)[] = []
private dataListeners: ((payload: PtyDataEvent) => void)[] = []
private exitListeners: ((payload: { id: string; code: number }) => void)[] = []
constructor(opts: {
@ -148,6 +145,10 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
await this.providerFor(id).clearBuffer(id)
}
async closeStartupQueryAuthority(id: string): Promise<number> {
return (await this.providerFor(id).closeStartupQueryAuthority?.(id)) ?? 0
}
acknowledgeDataEvent(id: string, charCount: number): void {
this.providerFor(id).acknowledgeDataEvent(id, charCount)
}
@ -187,9 +188,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
return this.fallback.getProfiles()
}
onData(
callback: (payload: { id: string; data: string; sequenceChars?: number }) => void
): () => void {
onData(callback: (payload: PtyDataEvent) => void): () => void {
this.dataListeners.push(callback)
return () => {
const idx = this.dataListeners.indexOf(callback)

View File

@ -106,6 +106,11 @@ describe('Session', () => {
cols?: number
rows?: number
launchAgent?: TuiAgent
startupIngress?: {
colors: { foreground: string; background: string }
deadlineMs: number
echoProjection?: 'windows-conpty-esc-stripped'
}
wslDistro?: string
}): Session {
session = new Session({
@ -116,6 +121,7 @@ describe('Session', () => {
wslDistro: opts?.wslDistro,
subprocess,
shellReadySupported: opts?.shellReadySupported ?? false,
...(opts?.startupIngress ? { startupIngress: opts.startupIngress } : {}),
...(opts?.shellReadyTimeoutMs !== undefined
? { shellReadyTimeoutMs: opts.shellReadyTimeoutMs }
: {})
@ -187,6 +193,56 @@ describe('Session', () => {
expect(received1).toEqual(['broadcast'])
expect(received2).toEqual(['broadcast'])
})
it('classifies startup queries and cooked echoes before model, persistence, and fanout', () => {
createSession({
startupIngress: {
colors: { foreground: '#2e3434', background: '#ffffff' },
deadlineMs: 5_000,
echoProjection: 'windows-conpty-esc-stripped'
}
})
const onData = vi.fn()
session.attachClient({ onData, onExit: () => {} })
const query = '\x1b]10;?\x07'
const echo = ']10;rgb:2e2e/3434/3434\\'
subprocess.simulateData(query)
subprocess.simulateData(echo)
subprocess.simulateData('prompt')
expect(subprocess.written).toEqual(['\x1b]10;rgb:2e2e/3434/3434\x1b\\'])
expect(onData.mock.calls).toEqual([
['', query.length, true, query.length],
['', echo.length, true, query.length + echo.length],
['prompt']
])
expect(session.takePendingOutput(false)?.records).toEqual([
{ kind: 'output', data: 'prompt' }
])
expect(session.getSnapshot()).toMatchObject({
outputSequence: query.length + echo.length + 'prompt'.length
})
expect(session.getSnapshot()?.snapshotAnsi).toContain('prompt')
expect(session.getSnapshot()?.snapshotAnsi).not.toContain(']10;rgb')
})
it('releases a held cooked-echo prefix before taking a snapshot', () => {
createSession({
startupIngress: {
colors: { foreground: '#2e3434', background: '#ffffff' },
deadlineMs: 5_000,
echoProjection: 'windows-conpty-esc-stripped'
}
})
subprocess.simulateData('\x1b]10;?\x07')
subprocess.simulateData(']10;rgb:2e2e/')
const snapshot = session.getSnapshot()
expect(snapshot?.snapshotAnsi).toContain(']10;rgb:2e2e/')
expect(snapshot?.outputSequence).toBe('\x1b]10;?\x07]10;rgb:2e2e/'.length)
})
})
describe('write', () => {

View File

@ -12,6 +12,11 @@ import { isPowerShellProcess } from '../../shared/shell-process-detection'
import { killWithDescendantSweep } from '../pty-descendant-termination'
import type { TuiAgent } from '../../shared/types'
import { PhysicalExitTracker } from '../../shared/physical-exit-tracker'
import {
PtyStartupIngress,
type PtyIngressEmission,
type PtyStartupIngressIntent
} from '../../shared/pty-startup-ingress'
import type {
PendingOutputRecord,
SessionState,
@ -97,11 +102,12 @@ export type SessionOptions = {
// reaper, dead sessions (and their ~5000-row scrollback emulators) accumulate
// for the lifetime of the long-lived daemon process.
onExit?: (code: number) => void
startupIngress?: PtyStartupIngressIntent
}
type AttachedClient = {
token: symbol
onData: (data: string) => void
onData: (data: string, rawLength?: number, transformed?: boolean, seq?: number) => void
onExit: (code: number) => void
}
@ -135,6 +141,7 @@ export class Session {
private forceKillSent = false
private subprocessDisposed = false
private readonly physicalExit = new PhysicalExitTracker()
private readonly startupIngress: PtyStartupIngress
constructor(opts: SessionOptions) {
this.sessionId = opts.sessionId
@ -170,6 +177,11 @@ export class Session {
}
this.postReadyFlushGate = new PostReadyFlushGate(() => this.flushPreReadyQueue())
this.startupIngress = new PtyStartupIngress({
...(opts.startupIngress ? { intent: opts.startupIngress } : {}),
write: (data) => this.subprocess.write(data),
onEmission: (emission) => this.emitSubprocessOutput(emission)
})
this.subprocess.onData((data) => this.handleSubprocessData(data))
this.subprocess.onExit((code) => this.handleSubprocessExit(code))
}
@ -389,7 +401,7 @@ export class Session {
this.subprocess.signal(sig)
}
attachClient(client: { onData: (data: string) => void; onExit: (code: number) => void }): symbol {
attachClient(client: Omit<AttachedClient, 'token'>): symbol {
const token = Symbol('attach')
this.attachedClients.push({ token, ...client })
return token
@ -413,6 +425,7 @@ export class Session {
}
getSnapshot(opts: { scrollbackRows?: number } = {}): TerminalSnapshot | null {
this.startupIngress.snapshotBarrier()
if (this._disposed) {
return null
}
@ -518,7 +531,9 @@ export class Session {
}
prepareForFinalSnapshot(): string {
return this.releaseHeldShellReadyBytes()
const held = this.releaseHeldShellReadyBytes()
this.startupIngress.snapshotBarrier()
return held
}
dispose(): void {
@ -533,6 +548,8 @@ export class Session {
// move this capture below #teardownSubprocess or the `_state = 'exited'`
// assignment — #teardownSubprocess flips `_disposed` but the invariant
// depends on the PRE-flip value of `_state`.
this.releaseHeldShellReadyBytes()
this.startupIngress.drainAndClose()
const wasTerminating = this._isTerminating && this._state !== 'exited'
const clientsToNotify = wasTerminating ? this.attachedClients.slice() : []
if (wasTerminating) {
@ -667,25 +684,29 @@ export class Session {
this.postReadyFlushGate.notifyData()
}
this.emitSubprocessOutput(data)
this.startupIngress.accept(data)
}
private emitSubprocessOutput(data: string): void {
if (data.length === 0) {
return
}
private emitSubprocessOutput(emission: PtyIngressEmission): void {
const { data } = emission
const rawLength = emission.rawEndSeq - emission.rawStartSeq
// Why: daemon stream thinning can omit bytes before main sees them. The
// absolute count lets an authoritative snapshot cover those gaps while
// renderer reconciliation deduplicates any queued post-snapshot tail.
this.outputSequence += data.length
this.outputSequence += rawLength
// Feed data to headless emulator for state tracking
this.emulator.write(data)
this.recordPendingOutput({ kind: 'output', data })
if (data.length > 0) {
this.emulator.write(data)
this.recordPendingOutput({ kind: 'output', data })
}
// Broadcast to attached clients
for (const client of this.attachedClients) {
client.onData(data)
if (emission.transformed || rawLength !== data.length) {
client.onData(data, rawLength, true, this.outputSequence)
} else {
client.onData(data)
}
}
}
@ -695,13 +716,14 @@ export class Session {
return
}
this.releaseHeldShellReadyBytes()
this.startupIngress.drainAndClose()
this._exitCode = code
this._state = 'exited'
this._isTerminating = false
// Why resume:false — the child is reaped, so there is nothing to unblock;
// only the failsafe timer must not outlive the session.
this.releaseProducerPause({ resume: false })
this.releaseHeldShellReadyBytes()
if (this.killTimer) {
clearTimeout(this.killTimer)
@ -740,10 +762,14 @@ export class Session {
// bytes can be stripped. If readiness never completes, preserve the
// previous behavior by releasing any held prefix before timeout or exit
// state changes discard it.
this.emitSubprocessOutput(heldBytes)
this.startupIngress.accept(heldBytes)
return heldBytes
}
closeStartupQueryAuthority(): number {
return this.startupIngress.closeQueryAuthority()
}
private transitionToReady(postMarkerBytesObserved = false): void {
this._shellState = 'ready'
this.shellReadyScanState = null

View File

@ -1,6 +1,7 @@
import type { StartupCommandDelivery } from '../../shared/codex-startup-delivery'
import type { TuiAgent } from '../../shared/types'
import type { ShellReadyState, TerminalSnapshot } from './types'
import type { PtyStartupIngressIntent } from '../../shared/pty-startup-ingress'
export type CreateOrAttachOptions = {
sessionId: string
@ -19,7 +20,11 @@ export type CreateOrAttachOptions = {
shellReadySupported?: boolean
shellReadyTimeoutMs?: number
historySeed?: string
streamClient: { onData: (data: string) => void; onExit: (code: number) => void }
startupIngress?: PtyStartupIngressIntent
streamClient: {
onData: (data: string, rawLength?: number, transformed?: boolean, seq?: number) => void
onExit: (code: number) => void
}
}
export type CreateOrAttachResult = {

View File

@ -0,0 +1,27 @@
import type { StartupCommandDelivery } from '../../shared/codex-startup-delivery'
import type { SubprocessHandle } from './session'
import type { TakePendingOutputResult, TerminalSnapshot } from './types'
export type TerminalHostOptions = {
spawnSubprocess: (opts: {
sessionId: string
cols: number
rows: number
cwd?: string
env?: Record<string, string>
envToDelete?: string[]
command?: string
startupCommandDelivery?: StartupCommandDelivery
shellOverride?: string
terminalWindowsWslDistro?: string | null
terminalWindowsPowerShellImplementation?: 'auto' | 'powershell.exe' | 'pwsh.exe'
}) => SubprocessHandle
// Why: graceful shutdown checkpoints must finish in-process before teardown.
onFinalCheckpoint?: (
sessionId: string,
snapshot: TerminalSnapshot,
records: TakePendingOutputResult['records']
) => void
// Why: tests need deterministic tombstone eviction without thousands of sessions.
maxTombstones?: number
}

View File

@ -1,8 +1,7 @@
import { Session, type SubprocessHandle } from './session'
import { Session } from './session'
import { normalizePtySize } from './daemon-pty-size'
import { shellPathSupportsPtyStartupBarrier } from './shell-ready'
import { resolveProcessCwd } from '../providers/process-cwd'
import type { StartupCommandDelivery } from '../../shared/codex-startup-delivery'
import { buildStartupCommandSubmission } from '../../shared/startup-command-submission'
import {
SessionNotFoundError,
@ -11,42 +10,17 @@ import {
type TerminalSnapshot
} from './types'
import type { CreateOrAttachOptions, CreateOrAttachResult } from './terminal-host-create-contract'
import type { TerminalHostOptions } from './terminal-host-options'
import { shutdownTerminalHostSessions } from './terminal-host-session-shutdown'
import { TerminalSessionTeardown } from './terminal-session-teardown'
import { resolveWslSessionContext } from './wsl-session-context'
import { getDaemonSessionResultMetadata } from './daemon-create-or-attach-result'
export type { CreateOrAttachOptions, CreateOrAttachResult } from './terminal-host-create-contract'
export type { TerminalHostOptions } from './terminal-host-options'
const DEFAULT_MAX_TOMBSTONES = 1000
export type TerminalHostOptions = {
spawnSubprocess: (opts: {
sessionId: string
cols: number
rows: number
cwd?: string
env?: Record<string, string>
envToDelete?: string[]
command?: string
startupCommandDelivery?: StartupCommandDelivery
shellOverride?: string
terminalWindowsWslDistro?: string | null
terminalWindowsPowerShellImplementation?: 'auto' | 'powershell.exe' | 'pwsh.exe'
}) => SubprocessHandle
// Why: on graceful shutdown, the host writes final checkpoints for all live
// sessions before killing them. This bypasses the RPC round-trip — the daemon
// writes checkpoints in-process, guaranteeing completion before teardown.
onFinalCheckpoint?: (
sessionId: string,
snapshot: TerminalSnapshot,
records: TakePendingOutputResult['records']
) => void
// Why: production keeps a large cap, but tests need a small deterministic cap
// without spawning thousands of full terminal sessions.
maxTombstones?: number
}
export class TerminalHost {
private sessions = new Map<string, Session>()
private sessionTeardown = new TerminalSessionTeardown(this.sessions)
@ -147,6 +121,7 @@ export class TerminalHost {
subprocess,
shellReadySupported,
historySeed: opts.historySeed,
...(opts.startupIngress ? { startupIngress: opts.startupIngress } : {}),
wslDistro,
// Why: reap the dead session (dispose emulator + drop from the map) the
// moment its subprocess exits, instead of retaining it for the daemon's
@ -197,6 +172,10 @@ export class TerminalHost {
this.getAliveSession(sessionId).write(data)
}
closeStartupQueryAuthority(sessionId: string): number {
return this.getAliveSession(sessionId).closeStartupQueryAuthority()
}
resize(sessionId: string, cols: number, rows: number): void {
this.getAliveSession(sessionId).resize(cols, rows)
}

View File

@ -0,0 +1,11 @@
export type TerminalModes = {
bracketedPaste: boolean
mouseTracking: boolean
mouseTrackingMode?: 'none' | 'x10' | 'vt200' | 'drag' | 'any'
sgrMouseMode?: boolean
sgrMousePixelsMode?: boolean
applicationCursor: boolean
alternateScreen: boolean
/** Kitty keyboard protocol flags used only to reseed a warm daemon emulator. */
kittyKeyboardFlags?: number
}

View File

@ -0,0 +1,20 @@
import type { TerminalOscLinkRange } from '../../shared/terminal-osc-link-ranges'
import type { TerminalModes } from './terminal-modes'
export type TerminalSnapshot = {
snapshotAnsi: string
/** Parser tail is already counted by the snapshot sequence and must restore last. */
pendingEscapeTailAnsi?: string
/** Normal buffer captured separately while snapshotAnsi holds an alternate buffer. */
scrollbackAnsi: string
oscLinks?: TerminalOscLinkRange[]
rehydrateSequences: string
cwd: string | null
modes: TerminalModes
cols: number
rows: number
scrollbackLines: number
lastTitle?: string
/** Optional because persisted snapshots and older v19 daemons lack it. */
outputSequence?: number
}

View File

@ -1,4 +1,3 @@
import type { TerminalOscLinkRange } from '../../shared/terminal-osc-link-ranges'
import type {
ConfirmForegroundProcessRequest,
GetForegroundProcessRequest
@ -12,67 +11,24 @@ export type {
// ─── Protocol Version ────────────────────────────────────────────────
import type { StartupCommandDelivery } from '../../shared/codex-startup-delivery'
import type { TuiAgent } from '../../shared/types'
// Why: daemons can survive app updates. Bump for IPC wire-shape changes, or
// when daemon-baked behavior cannot be delivered by on-disk wrapper refresh.
// Why: bump when adding daemon wire behavior so same-version old daemons do
// not silently accept the handshake and then reject new RPCs.
export const PROTOCOL_VERSION = 24
export const GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION = 22
export const CLEAN_DISCONNECT_PROTOCOL_VERSION = 24
export const PREVIOUS_DAEMON_PROTOCOL_VERSIONS = [
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23
] as const
import type { PtyStartupIngressIntent } from '../../shared/pty-startup-ingress'
export type { TerminalModes } from './terminal-modes'
import type { TerminalSnapshot } from './terminal-snapshot'
export type { TerminalSnapshot } from './terminal-snapshot'
export {
CLEAN_DISCONNECT_PROTOCOL_VERSION,
GIT_CREDENTIAL_GUARD_HOST_PROTOCOL_VERSION,
PREVIOUS_DAEMON_PROTOCOL_VERSIONS,
PROTOCOL_VERSION,
PTY_STARTUP_INGRESS_PROTOCOL_VERSION,
supportsPtyStartupIngress
} from './daemon-protocol-version'
// ─── Session State Machine ──────────────────────────────────────────
export type SessionState = 'created' | 'spawning' | 'running' | 'exiting' | 'exited'
export type ShellReadyState = 'pending' | 'ready' | 'timed_out' | 'unsupported'
// ─── Terminal Snapshot ──────────────────────────────────────────────
export type TerminalSnapshot = {
snapshotAnsi: string
/** Trailing incomplete escape sequence the emulator ingested but xterm's
* parser is still holding (a PTY read ended mid-escape). Restorers must
* write this LAST after their own post-replay resets, immediately before
* post-snapshot live chunks so the continuation bytes complete it
* exactly as live (Bug E / #7329, notes/garble-fuzz-divergences.md). Its
* bytes are already counted by the snapshot seq. */
pendingEscapeTailAnsi?: string
/** Normal buffer captured separately while snapshotAnsi holds an active
* alternate buffer. Empty for normal-screen snapshots. */
scrollbackAnsi: string
oscLinks?: TerminalOscLinkRange[]
rehydrateSequences: string
cwd: string | null
modes: TerminalModes
cols: number
rows: number
scrollbackLines: number
lastTitle?: string
/** Absolute UTF-16 character count ingested by this live daemon session.
* Optional because persisted snapshots and older v19 daemons lack it. */
outputSequence?: number
}
export type TerminalModes = {
bracketedPaste: boolean
mouseTracking: boolean
mouseTrackingMode?: 'none' | 'x10' | 'vt200' | 'drag' | 'any'
sgrMouseMode?: boolean
sgrMousePixelsMode?: boolean
applicationCursor: boolean
alternateScreen: boolean
/** Kitty keyboard protocol flags (CSI = u pushes) for emulator re-seed
* parity ONLY. Consumed by the daemon warm-reattach path: the spawn
* result threads them into seedHeadlessTerminal, which re-applies them to
* the fresh runtime emulator (HeadlessEmulator.applyKittyKeyboardFlags)
* so hidden `CSI ? u` answers the real flags instead of ?0u.
* rehydrateSequences must never push these into a renderer xterm
* POST_REPLAY_REATTACH_RESET's deliberate kitty reset stays authoritative
* (terminal-query-authority.md §kitty). */
kittyKeyboardFlags?: number
}
// The on-disk checkpoint.json shape lives in daemon-checkpoint-file.ts (it
// depends only on TerminalModes here) — re-exported so existing importers of
// `./types` keep working.
@ -115,9 +71,16 @@ export type CreateOrAttachRequest = {
shellReadyTimeoutMs?: number
/** Recovered ANSI applied before the new subprocess can emit startup output. */
historySeed?: string
startupIngress?: PtyStartupIngressIntent
}
}
export type CloseStartupQueryAuthorityRequest = {
id: string
type: 'closeStartupQueryAuthority'
payload: { sessionId: string }
}
export type CancelCreateOrAttachRequest = {
id: string
type: 'cancelCreateOrAttach'
@ -337,6 +300,7 @@ export type DaemonRequest =
| GetSnapshotRequest
| GetSizeRequest
| TakePendingOutputRequest
| CloseStartupQueryAuthorityRequest
// ─── RPC Responses (Daemon → Client, on control socket) ────────────

View File

@ -3852,7 +3852,8 @@ describe('registerPtyHandlers', () => {
result.id,
'daemon output',
expect.any(Number),
'daemon output'.length
'daemon output'.length,
undefined
)
expect(mainWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
id: result.id,
@ -8257,14 +8258,17 @@ describe('registerPtyHandlers', () => {
mockProc.proc.write.mockClear()
mainWindow.webContents.send.mockClear()
mockProc.emitData('\x1b]10;?\x1b\\\x1b]11;?\x1b\\ready')
const sourceData = '\x1b]10;?\x1b\\\x1b]11;?\x1b\\ready'
mockProc.emitData(sourceData)
expect(mockProc.proc.write).toHaveBeenCalledWith('\x1b]10;rgb:eeee/eeee/eeee\x1b\\')
expect(mockProc.proc.write).toHaveBeenCalledWith('\x1b]11;rgb:1111/1111/1111\x1b\\')
vi.advanceTimersByTime(2)
expect(mainWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
id: spawnResult.id,
data: 'ready'
data: 'ready',
rawLength: sourceData.length,
transformed: true
})
} finally {
vi.useRealTimers()
@ -8291,14 +8295,17 @@ describe('registerPtyHandlers', () => {
mockProc.proc.write.mockClear()
mainWindow.webContents.send.mockClear()
mockProc.emitData('\x1b]10;?;?\x1b\\ready')
const sourceData = '\x1b]10;?;?\x1b\\ready'
mockProc.emitData(sourceData)
expect(mockProc.proc.write).toHaveBeenCalledWith('\x1b]10;rgb:eeee/eeee/eeee\x1b\\')
expect(mockProc.proc.write).toHaveBeenCalledWith('\x1b]11;rgb:1111/1111/1111\x1b\\')
vi.advanceTimersByTime(2)
expect(mainWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
id: spawnResult.id,
data: 'ready'
data: 'ready',
rawLength: sourceData.length,
transformed: true
})
} finally {
vi.useRealTimers()
@ -8371,13 +8378,28 @@ describe('registerPtyHandlers', () => {
}
})
it('answers daemon agent startup OSC color queries before spawn resolves', async () => {
it('accepts source-classified daemon startup spans before spawn resolves', async () => {
vi.useFakeTimers()
let dataHandler: ((payload: { id: string; data: string }) => void) | null = null
type ProviderData = {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}
let dataHandler: ((payload: ProviderData) => void) | null = null
const write = vi.fn()
const spawn = vi.fn(async (options: { sessionId?: string }) => {
const query = '\x1b]10;?\x1b\\\x1b]11;?\x1b\\'
const spawn = vi.fn(async (options: { sessionId?: string; startupIngress?: unknown }) => {
const id = options.sessionId ?? 'daemon-pty'
dataHandler?.({ id, data: '\x1b]10;?\x1b\\\x1b]11;?\x1b\\daemon-ready' })
dataHandler?.({
id,
data: '',
sequenceChars: query.length,
transformed: true,
seq: query.length
})
dataHandler?.({ id, data: 'daemon-ready' })
return { id }
})
setLocalPtyProvider({
@ -8395,7 +8417,7 @@ describe('registerPtyHandlers', () => {
getForegroundProcess: vi.fn(),
serialize: vi.fn(),
revive: vi.fn(),
onData: vi.fn((handler: (payload: { id: string; data: string }) => void) => {
onData: vi.fn((handler: (payload: ProviderData) => void) => {
dataHandler = handler
return () => {}
}),
@ -8408,7 +8430,16 @@ describe('registerPtyHandlers', () => {
} as never)
try {
registerPtyHandlers(mainWindow as never)
let seq = 0
const runtime = {
setPtyController: vi.fn(),
createPreAllocatedTerminalHandle: vi.fn(() => null),
onPtyData: vi.fn(
(_id: string, _data: string, _at: number, rawLength: number) => (seq += rawLength)
),
registerPty: vi.fn()
}
registerPtyHandlers(mainWindow as never, runtime as never)
const spawnResult = (await handlers.get('pty:spawn')!(null, {
cols: 80,
rows: 24,
@ -8420,22 +8451,39 @@ describe('registerPtyHandlers', () => {
}
})) as { id: string }
expect(write).toHaveBeenCalledWith(spawnResult.id, '\x1b]10;rgb:eeee/eeee/eeee\x1b\\')
expect(write).toHaveBeenCalledWith(spawnResult.id, '\x1b]11;rgb:1111/1111/1111\x1b\\')
expect(spawn).toHaveBeenCalledWith(
expect.objectContaining({
startupIngress: expect.objectContaining({
colors: { foreground: '#eeeeee', background: '#111111' },
deadlineMs: 5_000
})
})
)
expect(write).not.toHaveBeenCalled()
vi.advanceTimersByTime(2)
expect(mainWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
id: spawnResult.id,
data: 'daemon-ready'
data: 'daemon-ready',
seq: query.length + 'daemon-ready'.length,
rawLength: query.length + 'daemon-ready'.length,
transformed: true
})
} finally {
vi.useRealTimers()
}
})
it('drops renderer sequence metadata when an answered OSC query is batched', async () => {
it('preserves source raw sequence metadata when a consumed query is batched', async () => {
vi.useFakeTimers()
type ProviderData = {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}
const providerEvents: {
dataHandler?: (payload: { id: string; data: string }) => void
dataHandler?: (payload: ProviderData) => void
} = {}
const write = vi.fn()
const spawn = vi.fn(async (options: { sessionId?: string }) => ({
@ -8456,7 +8504,7 @@ describe('registerPtyHandlers', () => {
getForegroundProcess: vi.fn(),
serialize: vi.fn(),
revive: vi.fn(),
onData: vi.fn((handler: (payload: { id: string; data: string }) => void) => {
onData: vi.fn((handler: (payload: ProviderData) => void) => {
providerEvents.dataHandler = handler
return () => {}
}),
@ -8471,8 +8519,8 @@ describe('registerPtyHandlers', () => {
const runtime = {
setPtyController: vi.fn(),
createPreAllocatedTerminalHandle: vi.fn(() => null),
onPtyData: vi.fn((_id: string, data: string) => {
seq += data.length
onPtyData: vi.fn((_id: string, data: string, _at: number, rawLength = data.length) => {
seq += rawLength
return seq
}),
registerPty: vi.fn()
@ -8493,17 +8541,24 @@ describe('registerPtyHandlers', () => {
mainWindow.webContents.send.mockClear()
providerEvents.dataHandler?.({ id: spawnResult.id, data: 'prefix' })
const query = '\x1b]10;?\x1b\\\x1b]11;?\x1b\\'
providerEvents.dataHandler?.({
id: spawnResult.id,
data: '\x1b]10;?\x1b\\\x1b]11;?\x1b\\ready'
data: '',
sequenceChars: query.length,
transformed: true,
seq: 'prefix'.length + query.length
})
providerEvents.dataHandler?.({ id: spawnResult.id, data: 'ready' })
vi.advanceTimersByTime(2)
expect(write).toHaveBeenCalledWith(spawnResult.id, '\x1b]10;rgb:eeee/eeee/eeee\x1b\\')
expect(write).toHaveBeenCalledWith(spawnResult.id, '\x1b]11;rgb:1111/1111/1111\x1b\\')
expect(write).not.toHaveBeenCalled()
expect(mainWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
id: spawnResult.id,
data: 'prefixready'
data: 'prefixready',
seq: 'prefix'.length + query.length + 'ready'.length,
rawLength: 'prefix'.length + query.length + 'ready'.length,
transformed: true
})
} finally {
vi.useRealTimers()
@ -9925,7 +9980,8 @@ describe('registerPtyHandlers', () => {
result.id,
'hidden output',
expect.any(Number),
'hidden output'.length
'hidden output'.length,
undefined
)
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(1)
// Why out-of-band: an in-band empty pty:data chunk is ambiguous with
@ -10477,7 +10533,8 @@ describe('registerPtyHandlers', () => {
'daemon-session',
'pre-spawn prompt\x1b[c',
expect.any(Number),
'pre-spawn prompt\x1b[c'.length
'pre-spawn prompt\x1b[c'.length,
undefined
)
expect(mainWindow.webContents.send).toHaveBeenCalledTimes(1)
expect(mainWindow.webContents.send).toHaveBeenCalledWith('pty:modelRestoreNeeded', {

View File

@ -149,13 +149,7 @@ import { isHostCodexHomeForWsl, isWslCodexHomeForHost } from '../pty/codex-home-
import { buildConfiguredProxyEnv, type NetworkProxySettings } from '../../shared/network-proxy'
import { resolveSetupAgentSequenceLaunchCommand } from '../../shared/setup-agent-sequencing'
import { parseWorkspaceKey } from '../../shared/workspace-scope'
import {
answerStartupTerminalColorQueries,
clearStartupTerminalColorQueryReplies,
getStartupTerminalColorQueryReplyColors,
moveStartupTerminalColorQueryReplies,
registerStartupTerminalColorQueryReplies
} from './terminal-startup-color-query-replies'
import { getStartupTerminalColorQueryReplyColors } from './terminal-startup-color-query-replies'
import {
assertFolderWorkspacePathUsable,
getFolderWorkspacePathStatus
@ -459,20 +453,14 @@ function tryGetProviderForPty(ptyId: string): IPtyProvider | undefined {
}
}
function getProviderForStartupTerminalColorReply(ptyId: string): IPtyProvider | undefined {
const ownedConnectionId = ptyOwnership.get(ptyId)
if (ownedConnectionId !== undefined) {
return getProvider(ownedConnectionId)
function closeStartupQueryAuthorityForPty(ptyId: string): void {
try {
void Promise.resolve(tryGetProviderForPty(ptyId)?.closeStartupQueryAuthority?.(ptyId)).catch(
() => {}
)
} catch {
/* Best-effort handoff; the bounded source deadline remains the fallback. */
}
const parsedSshId = parseAppSshPtyId(ptyId)
if (parsedSshId) {
return getProvider(parsedSshId.connectionId)
}
return localProvider
}
export function answerStartupTerminalColorQueriesForPty(ptyId: string, data: string): string {
return answerStartupTerminalColorQueries(ptyId, data, getProviderForStartupTerminalColorReply)
}
function normalizeNodePtySpawnError(err: unknown): Error {
@ -1189,7 +1177,6 @@ export function clearProviderPtyState(id: string): void {
rendererVisibilityKnownPtys.delete(id)
pendingHiddenRendererResizeOutputPtys.delete(id)
deliveredHiddenRendererResizeOutputPtys.delete(id)
clearStartupTerminalColorQueryReplies(id)
// Why: every PTY teardown path funnels through here (local exit, daemon
// shutdown, SSH exit/connection teardown) — hidden/interest gate bits must
// not outlive the PTY or a reused map entry could silently gate a new one.
@ -1637,7 +1624,8 @@ export function registerPtyHandlers(
markClaudePtyExited(id)
runtime?.onPtyExit(id, code)
},
onData: (id, data, timestamp) => runtime?.onPtyData(id, data, timestamp)
onData: (id, data, timestamp, sequenceChars, transformed) =>
runtime?.onPtyData(id, data, timestamp, sequenceChars ?? data.length, transformed)
})
}
@ -1648,6 +1636,8 @@ export function registerPtyHandlers(
type PendingPtyData = {
data: string
startSeq?: number
rawLength?: number
transformed?: true
containsBackgroundOutput?: boolean
// Why droppedOutput (not main's droppedBacklog trim): this branch bounds
// the unsent backlog with the O(1) drop-to-sentinel + query-salvage +
@ -1662,6 +1652,7 @@ export function registerPtyHandlers(
data: string
seq?: number
rawLength?: number
transformed?: boolean
background?: boolean
droppedOutput?: boolean
}
@ -2057,20 +2048,23 @@ export function registerPtyHandlers(
return true
}
function getChunkStartSeq(endSeq: number | undefined, data: string): number | undefined {
return typeof endSeq === 'number' ? Math.max(0, endSeq - data.length) : undefined
}
function makePtyDataPayload(
id: string,
data: string,
startSeq: number | undefined,
containsBackgroundOutput: boolean | undefined
containsBackgroundOutput: boolean | undefined,
rawLength = data.length,
transformed = false
): PtyDataPayload {
const payload: PtyDataPayload = { id, data }
if (typeof startSeq === 'number') {
payload.seq = startSeq + data.length
payload.rawLength = data.length
payload.seq = startSeq + rawLength
}
if (typeof startSeq === 'number' || rawLength !== data.length || transformed) {
payload.rawLength = rawLength
}
if (transformed) {
payload.transformed = true
}
if (containsBackgroundOutput === true) {
payload.background = true
@ -2362,7 +2356,9 @@ export function registerPtyHandlers(
data: string,
startSeq: number | undefined,
preservesSeq: boolean,
containsBackgroundOutput: boolean
containsBackgroundOutput: boolean,
rawLength = data.length,
transformed = false
): PendingPtyData {
// Why: once over the cap, stay dropped at O(1) memory until the renderer
// can receive again — the restore sentinel supersedes any interim bytes.
@ -2377,21 +2373,21 @@ export function registerPtyHandlers(
}
const nextContainsBackgroundOutput =
existing?.containsBackgroundOutput === true || containsBackgroundOutput
if (!preservesSeq) {
return dropOversizedPendingPtyData(id, {
data: (existing?.data ?? '') + data,
...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {})
})
}
if (!existing) {
return dropOversizedPendingPtyData(id, {
data,
...(typeof startSeq === 'number' ? { startSeq } : {}),
...(rawLength !== data.length ? { rawLength } : {}),
...(transformed ? { transformed: true } : {}),
...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {})
})
}
const existingRawLength = existing.rawLength ?? existing.data.length
const next: PendingPtyData = {
data: existing.data + data,
...(!preservesSeq || existing.transformed || transformed
? { rawLength: existingRawLength + rawLength, transformed: true as const }
: {}),
...(nextContainsBackgroundOutput ? { containsBackgroundOutput: true } : {})
}
if (typeof existing.startSeq === 'number') {
@ -2490,8 +2486,9 @@ export function registerPtyHandlers(
continue
}
const { data } = pending
const chunk = data.slice(0, PTY_BATCH_FLUSH_CHUNK_CHARS)
const remaining = data.slice(PTY_BATCH_FLUSH_CHUNK_CHARS)
const indivisible = pending.transformed === true
const chunk = indivisible ? data : data.slice(0, PTY_BATCH_FLUSH_CHUNK_CHARS)
const remaining = indivisible ? '' : data.slice(PTY_BATCH_FLUSH_CHUNK_CHARS)
if (remaining) {
const nextPending: PendingPtyData = { data: remaining }
if (typeof pending.startSeq === 'number') {
@ -2507,7 +2504,14 @@ export function registerPtyHandlers(
updateProducerFlowControl(id)
sendPtyDataToRenderer(
id,
makePtyDataPayload(id, chunk, pending.startSeq, pending.containsBackgroundOutput)
makePtyDataPayload(
id,
chunk,
pending.startSeq,
pending.containsBackgroundOutput,
pending.rawLength,
pending.transformed
)
)
writes++
}
@ -2580,7 +2584,9 @@ export function registerPtyHandlers(
payload.id,
remaining.data,
remaining.startSeq,
remaining.containsBackgroundOutput
remaining.containsBackgroundOutput,
remaining.rawLength,
remaining.transformed
)
)
}
@ -2668,19 +2674,14 @@ export function registerPtyHandlers(
const isLocalProvider = localProvider instanceof LocalPtyProvider
localDataUnsub = localProvider.onData((payload) => {
const rawLength = payload.sequenceChars ?? payload.data.length
const outputSeq = isLocalProvider
? runtime?.getPtyOutputSequence(payload.id)
: runtime?.onPtyData(
payload.id,
payload.data,
Date.now(),
payload.sequenceChars ?? payload.data.length
)
const rendererData = answerStartupTerminalColorQueriesForPty(payload.id, payload.data)
const preservesSeq =
rendererData === payload.data &&
(payload.sequenceChars === undefined || payload.sequenceChars === payload.data.length)
const startSeq = preservesSeq ? getChunkStartSeq(outputSeq, payload.data) : undefined
: runtime?.onPtyData(payload.id, payload.data, Date.now(), rawLength, payload.transformed)
const rendererData = payload.data
const preservesSeq = !payload.transformed && rawLength === payload.data.length
const startSeq =
typeof outputSeq === 'number' ? Math.max(0, outputSeq - rawLength) : undefined
if (mainWindow.isDestroyed()) {
// Why: clear the pending flush timer so it doesn't fire after the window
// is gone. Without this, macOS app re-activation leaks orphaned timers
@ -2713,7 +2714,7 @@ export function registerPtyHandlers(
}
return
}
if (rendererData.length === 0) {
if (rendererData.length === 0 && !payload.transformed) {
return
}
const containsBackgroundOutput =
@ -2728,7 +2729,9 @@ export function registerPtyHandlers(
rendererData,
startSeq,
preservesSeq,
containsBackgroundOutput
containsBackgroundOutput,
rawLength,
payload.transformed === true
)
const nextData = pending.data
const isInteractiveOutput = shouldSendInteractiveOutputNow(
@ -2760,8 +2763,12 @@ export function registerPtyHandlers(
id: payload.id,
data: nextData,
...(typeof pending.startSeq === 'number'
? { seq: pending.startSeq + nextData.length, rawLength: nextData.length }
? {
seq: pending.startSeq + (pending.rawLength ?? nextData.length),
rawLength: pending.rawLength ?? nextData.length
}
: {}),
...(pending.transformed ? { transformed: true } : {}),
...(pending.containsBackgroundOutput === true ? { background: true } : {}),
...(pending.droppedOutput === true ? { droppedOutput: true } : {})
})
@ -3904,10 +3911,6 @@ export function registerPtyHandlers(
})?.distro ?? null)
: null
const startupTerminalColorQueryReplyColors = getStartupTerminalColorQueryReplyColors(args)
const preSpawnStartupTerminalColorReplyPtyId =
startupTerminalColorQueryReplyColors && effectiveSessionId !== undefined
? (effectiveSessionAppId ?? effectiveSessionId)
: null
// Why: the renderer sets pane env for SSH too. Only forward it to the
// remote when the relay hook path is enabled; otherwise a newer relay
// could emit statuses this Orca build is not prepared to route.
@ -4015,6 +4018,11 @@ export function registerPtyHandlers(
const validatedLeafId = verifiedLeafId ?? metadataLeafId
let env: Record<string, string> | undefined = baseEnv
const effectiveShellOverride = terminalRuntimeOptions.shellOverride
const nativeWindowsConptySpawn = isNativeWindowsLocalPtySpawn({
connectionId: args.connectionId,
cwd: args.cwd,
shellOverride: effectiveShellOverride
})
const codexSelectionTarget = getCodexSelectionTargetForPty(
effectiveShellOverride,
cwd,
@ -4169,6 +4177,15 @@ export function registerPtyHandlers(
? (getSettings()?.terminalWindowsPowerShellImplementation ?? 'auto')
: undefined
}
if (startupTerminalColorQueryReplyColors) {
spawnOptions.startupIngress = {
colors: startupTerminalColorQueryReplyColors,
deadlineMs: 5_000,
...(nativeWindowsConptySpawn
? { echoProjection: 'windows-conpty-esc-stripped' as const }
: {})
}
}
const existingPaneSpawn = reservationPaneKey
? paneSpawnReservationsByPaneKey.get(reservationPaneKey)
: undefined
@ -4202,14 +4219,6 @@ export function registerPtyHandlers(
if (preAllocatedHandle) {
trustedTerminalHandleEnv.add(preAllocatedHandle)
}
if (preSpawnStartupTerminalColorReplyPtyId && startupTerminalColorQueryReplyColors) {
// Why: Codex probes OSC 10/11 with a 100 ms timeout and daemon PTYs
// can emit that query before spawn() resolves to the renderer.
registerStartupTerminalColorQueryReplies(
preSpawnStartupTerminalColorReplyPtyId,
startupTerminalColorQueryReplyColors
)
}
spawnTiming.mark('options')
const expectedPtyId = effectiveSessionAppId ?? effectiveSessionId
if (isDaemonHostSpawn && expectedPtyId) {
@ -4254,9 +4263,6 @@ export function registerPtyHandlers(
const spawnError = normalizeNodePtySpawnError(err)
const isIdentityMismatch =
isSshPtyIdentityMismatchError(spawnError) || isSshPtyIdentityMismatchError(rawMessage)
if (preSpawnStartupTerminalColorReplyPtyId) {
clearStartupTerminalColorQueryReplies(preSpawnStartupTerminalColorReplyPtyId)
}
if (effectiveSessionAppId !== undefined) {
if (isIdentityMismatch && hadSessionSizeBeforeAttach && sessionSizeBeforeAttach) {
ptySizes.set(effectiveSessionAppId, sessionSizeBeforeAttach)
@ -4337,33 +4343,14 @@ export function registerPtyHandlers(
// ownership, and a hidden-spawned agent should be paceable from its
// first flood, not from its first visibility transition.
syncPtyBackgroundedDelivery(result.id, 'spawn')
closeStartupQueryAuthorityForPty(result.id)
}
// Why: Phase-5 ConPTY DA1 — record the native-Windows-local-PTY
// determination from the spawn record before the headless seed below,
// so the runtime emulator's DA1 override exists from byte zero.
if (
isNativeWindowsLocalPtySpawn({
connectionId: args.connectionId,
cwd: args.cwd,
shellOverride: effectiveShellOverride
})
) {
if (nativeWindowsConptySpawn) {
markNativeWindowsConptyPty(result.id)
}
if (startupTerminalColorQueryReplyColors) {
if (result.isReattach) {
if (preSpawnStartupTerminalColorReplyPtyId) {
clearStartupTerminalColorQueryReplies(preSpawnStartupTerminalColorReplyPtyId)
}
} else if (preSpawnStartupTerminalColorReplyPtyId) {
moveStartupTerminalColorQueryReplies(preSpawnStartupTerminalColorReplyPtyId, result.id)
} else {
registerStartupTerminalColorQueryReplies(
result.id,
startupTerminalColorQueryReplyColors
)
}
}
const relayResultId = getRelayPtyId(args.connectionId, result.id)
if (store && args.connectionId) {
// Why: remote PTYs live in the SSH relay grace window after Orca
@ -5085,6 +5072,7 @@ export function registerPtyHandlers(
rendererVisibilityKnownPtys.add(args.id)
if (args.visible) {
visibleRendererPtys.add(args.id)
closeStartupQueryAuthorityForPty(args.id)
} else {
visibleRendererPtys.delete(args.id)
}
@ -5101,6 +5089,7 @@ export function registerPtyHandlers(
})
if (args.hidden === true) {
markHiddenRendererPty(args.id)
closeStartupQueryAuthorityForPty(args.id)
// Why: bytes already queued for a newly hidden PTY are model-owned
// state; drop them now instead of holding them under ACK starvation.
// Reveal restores from the snapshot.

View File

@ -161,7 +161,6 @@ vi.mock('./pty', () => ({
clearProviderPtyState: vi.fn(),
deletePtyOwnership: vi.fn(),
setPtyOwnership: vi.fn(),
answerStartupTerminalColorQueriesForPty: vi.fn((_id: string, data: string) => data),
getSshPtyProvider: vi.fn(),
getPtyIdsForConnection: vi.fn().mockReturnValue([]),
isRendererPtyOutputPaused: vi.fn().mockReturnValue(false)
@ -809,7 +808,13 @@ describe('SSH IPC handlers', () => {
onData?.({ id: 'remote-pty', data: 'hello' })
onExit?.({ id: 'remote-pty', code: 7 })
expect(runtime.onPtyData).toHaveBeenCalledWith('remote-pty', 'hello', expect.any(Number))
expect(runtime.onPtyData).toHaveBeenCalledWith(
'remote-pty',
'hello',
expect.any(Number),
'hello'.length,
undefined
)
expect(runtime.onPtyExit).toHaveBeenCalledWith('remote-pty', 7)
})
@ -1180,7 +1185,13 @@ describe('SSH IPC handlers', () => {
targetId: 'ssh-1',
ports: expect.arrayContaining([expect.objectContaining({ port: 3000 })])
})
expect(secondRuntime.onPtyData).toHaveBeenCalledWith('remote-pty', 'hello', expect.any(Number))
expect(secondRuntime.onPtyData).toHaveBeenCalledWith(
'remote-pty',
'hello',
expect.any(Number),
'hello'.length,
undefined
)
expect(secondRuntime.onPtyExit).toHaveBeenCalledWith('remote-pty', 9)
expect(firstRuntime.onPtyData).not.toHaveBeenCalled()
expect(firstRuntime.onPtyExit).not.toHaveBeenCalled()

View File

@ -3,71 +3,10 @@ import { isTuiAgent } from '../../shared/tui-agent-config'
import { agentKindSchema } from '../../shared/telemetry-events'
import type { SleepingAgentLaunchConfig } from '../../shared/agent-session-resume'
import {
parseTerminalOscColorQuery,
terminalOscColorQueryReplies,
terminalOscColorQueryReply,
type TerminalOscColorQueryReplyColors,
type TerminalOscColorQuerySlot
type TerminalOscColorQueryReplyColors
} from '../../shared/terminal-osc-color-reply'
type StartupTerminalColorQueryProvider = {
write(id: string, data: string): void
}
type StartupTerminalColorQueryReplyState = {
colors: TerminalOscColorQueryReplyColors
pending: string
answeredSlots: Set<TerminalOscColorQuerySlot>
timeout: ReturnType<typeof setTimeout>
}
const STARTUP_TERMINAL_COLOR_QUERY_REPLY_WINDOW_MS = 5_000
const STARTUP_TERMINAL_COLOR_QUERY_PENDING_CHARS = 64
const startupTerminalColorQueryRepliesByPty = new Map<string, StartupTerminalColorQueryReplyState>()
export function clearStartupTerminalColorQueryReplies(ptyId: string): void {
const state = startupTerminalColorQueryRepliesByPty.get(ptyId)
if (!state) {
return
}
clearTimeout(state.timeout)
startupTerminalColorQueryRepliesByPty.delete(ptyId)
}
export function moveStartupTerminalColorQueryReplies(fromPtyId: string, toPtyId: string): void {
if (fromPtyId === toPtyId) {
return
}
const state = startupTerminalColorQueryRepliesByPty.get(fromPtyId)
if (!state) {
return
}
startupTerminalColorQueryRepliesByPty.delete(fromPtyId)
clearStartupTerminalColorQueryReplies(toPtyId)
startupTerminalColorQueryRepliesByPty.set(toPtyId, state)
}
export function registerStartupTerminalColorQueryReplies(
ptyId: string,
colors: TerminalOscColorQueryReplyColors
): void {
if (!terminalOscColorQueryReply(colors, 10) || !terminalOscColorQueryReply(colors, 11)) {
return
}
clearStartupTerminalColorQueryReplies(ptyId)
const timeout = setTimeout(
() => clearStartupTerminalColorQueryReplies(ptyId),
STARTUP_TERMINAL_COLOR_QUERY_REPLY_WINDOW_MS
)
timeout.unref?.()
startupTerminalColorQueryRepliesByPty.set(ptyId, {
colors,
pending: '',
answeredSlots: new Set(),
timeout
})
}
function normalizeTerminalColorQueryReplyColors(
value: unknown
): TerminalOscColorQueryReplyColors | null {
@ -117,81 +56,3 @@ export function getStartupTerminalColorQueryReplyColors(args: {
}
return normalizeTerminalColorQueryReplyColors(args.terminalColorQueryReplies)
}
function writeStartupTerminalColorQueryReplies(
ptyId: string,
slots: readonly TerminalOscColorQuerySlot[],
state: StartupTerminalColorQueryReplyState,
getProvider: (ptyId: string) => StartupTerminalColorQueryProvider | undefined
): boolean {
const replies = terminalOscColorQueryReplies(state.colors, slots)
let provider: StartupTerminalColorQueryProvider | undefined
try {
provider = replies ? getProvider(ptyId) : undefined
} catch {
provider = undefined
}
if (!replies || !provider) {
return false
}
try {
for (const [index, reply] of replies.entries()) {
const slot = slots[index]
if (slot === undefined) {
return false
}
provider.write(ptyId, reply)
state.answeredSlots.add(slot)
}
return true
} catch {
return false
}
}
export function answerStartupTerminalColorQueries(
ptyId: string,
data: string,
getProvider: (ptyId: string) => StartupTerminalColorQueryProvider | undefined
): string {
const state = startupTerminalColorQueryRepliesByPty.get(ptyId)
if (!state || data.length === 0) {
return data
}
const input = state.pending + data
let pending = ''
let output = ''
let offset = 0
while (offset < input.length) {
const candidateIndex = input.indexOf('\x1b', offset)
if (candidateIndex === -1) {
output += input.slice(offset)
break
}
output += input.slice(offset, candidateIndex)
const query = parseTerminalOscColorQuery(input, candidateIndex)
if (query.kind === 'none') {
output += input[candidateIndex]
offset = candidateIndex + 1
continue
}
if (query.kind === 'partial') {
const candidate = input.slice(candidateIndex)
if (candidate.length <= STARTUP_TERMINAL_COLOR_QUERY_PENDING_CHARS) {
pending = candidate
} else {
output += candidate
}
break
}
if (!writeStartupTerminalColorQueryReplies(ptyId, query.slots, state, getProvider)) {
output += input.slice(candidateIndex, query.endIndex)
}
offset = query.endIndex
}
state.pending = pending
if (state.answeredSlots.has(10) && state.answeredSlots.has(11)) {
clearStartupTerminalColorQueryReplies(ptyId)
}
return output
}

View File

@ -1582,6 +1582,47 @@ describe('LocalPtyProvider', () => {
expect(dataHandler).toHaveBeenCalledWith({ id, data: 'hello world' })
})
it('classifies startup queries before runtime and public data listeners', async () => {
const runtimeData = vi.fn()
const dataHandler = vi.fn()
provider.configure({ onData: runtimeData })
provider.onData(dataHandler)
const { id } = await provider.spawn({
cols: 80,
rows: 24,
startupIngress: {
colors: { foreground: '#2e3434', background: '#ffffff' },
deadlineMs: 5_000,
echoProjection: 'windows-conpty-esc-stripped'
}
})
const onDataCb = mockProc.onData.mock.calls[0][0]
const query = '\x1b]10;?\x07'
const echo = ']10;rgb:2e2e/3434/3434\\'
onDataCb(query)
onDataCb(echo)
onDataCb('prompt')
expect(mockProc.write).toHaveBeenCalledWith('\x1b]10;rgb:2e2e/3434/3434\x1b\\')
expect(runtimeData.mock.calls.map((call) => call.slice(1))).toEqual([
['', expect.any(Number), query.length, true],
['', expect.any(Number), echo.length, true],
['prompt', expect.any(Number)]
])
expect(dataHandler.mock.calls.map(([payload]) => payload)).toEqual([
{ id, data: '', sequenceChars: query.length, seq: query.length, transformed: true },
{
id,
data: '',
sequenceChars: echo.length,
seq: query.length + echo.length,
transformed: true
},
{ id, data: 'prompt' }
])
})
it('notifies exit listeners when PTY exits', async () => {
const exitHandler = vi.fn()
provider.onExit(exitHandler)

View File

@ -68,6 +68,7 @@ import { assertSafeAgentStartupCwd, resolveSafePtyDefaultCwd } from './pty-defau
import { ORCA_HERMES_STARTUP_QUERY_ENV } from '../../shared/hermes-startup-query'
import { PhysicalExitTracker } from '../../shared/physical-exit-tracker'
import { mergeGitConfigEnvProtocol } from '../../shared/git-credential-prompt-env'
import { PtyStartupIngress, type PtyIngressEmission } from '../../shared/pty-startup-ingress'
const PANE_IDENTITY_ENV_KEYS = [
'ORCA_PANE_KEY',
@ -122,11 +123,18 @@ export const LOCAL_PTY_FORCE_KILL_RETRY_MS = 250
let loadGeneration = 0
const ptyLoadGeneration = new Map<string, number>()
type DataCallback = (payload: { id: string; data: string }) => void
type DataCallback = (payload: {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}) => void
type ExitCallback = (payload: { id: string; code: number }) => void
const dataListeners = new Set<DataCallback>()
const exitListeners = new Set<ExitCallback>()
const startupIngressByPty = new Map<string, PtyStartupIngress>()
/**
* Returns a stable default cwd for locally spawned PTYs.
@ -492,7 +500,13 @@ export type LocalPtyProviderOptions = {
pwshAvailable?: () => boolean
onSpawned?: (id: string) => void
onExit?: (id: string, code: number) => void
onData?: (id: string, data: string, timestamp: number) => void
onData?: (
id: string,
data: string,
timestamp: number,
sequenceChars?: number,
transformed?: boolean
) => void
}
export class LocalPtyProvider implements IPtyProvider {
@ -907,6 +921,34 @@ export class LocalPtyProvider implements IPtyProvider {
ptyLoadGeneration.set(id, loadGeneration)
this.opts.onSpawned?.(id)
const emitIngressData = (emission: PtyIngressEmission): void => {
const sequenceChars = emission.rawEndSeq - emission.rawStartSeq
if (emission.transformed || sequenceChars !== emission.data.length) {
this.opts.onData?.(id, emission.data, Date.now(), sequenceChars, true)
} else {
this.opts.onData?.(id, emission.data, Date.now())
}
for (const cb of dataListeners) {
cb(
emission.transformed || sequenceChars !== emission.data.length
? {
id,
data: emission.data,
sequenceChars,
seq: emission.rawEndSeq,
transformed: true
}
: { id, data: emission.data }
)
}
}
const startupIngress = new PtyStartupIngress({
...(args.startupIngress ? { intent: args.startupIngress } : {}),
write: (data) => proc.write(data),
onEmission: emitIngressData
})
startupIngressByPty.set(id, startupIngress)
// Shell-ready startup command support
let resolveShellReady: ((signal: ShellReadySignal) => void) | null = null
let shellReadyTimeout: ReturnType<typeof setTimeout> | null = null
@ -938,10 +980,7 @@ export class LocalPtyProvider implements IPtyProvider {
if (heldBytes.length === 0) {
return
}
this.opts.onData?.(id, heldBytes, Date.now())
for (const cb of dataListeners) {
cb({ id, data: heldBytes })
}
startupIngress.accept(heldBytes)
}
if (args.command) {
if (shellReadyLaunch?.supportsReadyMarker) {
@ -977,13 +1016,7 @@ export class LocalPtyProvider implements IPtyProvider {
finishShellReady({ postMarkerBytesObserved: scanned.postMarkerBytesObserved })
}
}
if (data.length === 0) {
return
}
this.opts.onData?.(id, data, Date.now())
for (const cb of dataListeners) {
cb({ id, data })
}
startupIngress.accept(data)
})
if (onDataDisposable) {
disposables.push(onDataDisposable)
@ -1010,6 +1043,8 @@ export class LocalPtyProvider implements IPtyProvider {
}
startupCommandCleanup?.()
clearPtyState(id)
startupIngress.drainAndClose()
startupIngressByPty.delete(id)
// Why: release the master ptmx fd on the natural-exit path — without
// this, a shell that exits cleanly (the common case) never releases its
// fd until the next GC. See docs/fix-pty-fd-leak.md.
@ -1223,11 +1258,15 @@ export class LocalPtyProvider implements IPtyProvider {
// only safe at an empty prompt, and without a headless emulator this
// provider cannot tell whether input is pending.
try {
startupIngressByPty.get(id)?.snapshotBarrier()
ptyProcesses.get(id)?.clear()
} catch {
/* PTY may have just exited */
}
}
closeStartupQueryAuthority(id: string): number {
return startupIngressByPty.get(id)?.closeQueryAuthority() ?? 0
}
acknowledgeDataEvent(_id: string, _charCount: number): void {
/* no flow control for local */
}

View File

@ -0,0 +1,22 @@
import type { TerminalGitHubPRLink } from '../../shared/terminal-github-pr-link-detector'
export type PtyDataEvent = {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}
/** Notification-bearing fact a thinning transport detected while it held
* scan authority for a backgrounded PTY (see onBackgroundStreamEvent). */
export type PtyTransientFact =
| { kind: 'bell' }
| { kind: 'command-finished'; exitCode: number | null }
| { kind: 'pr-link'; link: TerminalGitHubPRLink }
| { kind: '2031-subscribe' }
export type PtyBackgroundStreamEvent =
| { id: string; kind: 'backgroundMarker'; background: boolean; scanSeedAnsi?: string }
| { id: string; kind: 'dataGap'; droppedChars: number; sequenceChars?: number }
| { id: string; kind: 'transientFact'; fact: PtyTransientFact }

View File

@ -49,6 +49,21 @@ describe('SshPtyProvider', () => {
expect(result).toEqual({ id: scopedPty1 })
})
it('gates fresh startup intent with the relay ingress capability version', async () => {
mux.request.mockResolvedValue({ id: 'pty-1' })
const startupIngress = {
colors: { foreground: '#eeeeee', background: '#111111' },
deadlineMs: 5_000
}
await provider.spawn({ cols: 80, rows: 24, startupIngress })
expect(mux.request).toHaveBeenCalledWith(
'pty.spawn',
expect.objectContaining({ startupIngressVersion: 1, startupIngress })
)
})
it('passes cwd and env through', async () => {
mux.request.mockResolvedValue({ id: 'pty-2' })
@ -324,6 +339,26 @@ describe('SshPtyProvider', () => {
})
})
it('never sends fresh startup intent on relay reattach', async () => {
mux.request.mockResolvedValue({ replay: 'restored' })
await provider.spawn({
cols: 80,
rows: 24,
sessionId: scopedPty1,
startupIngress: {
colors: { foreground: '#eeeeee', background: '#111111' },
deadlineMs: 5_000
}
})
expect(mux.request).toHaveBeenCalledWith(
'pty.attach',
expect.not.objectContaining({ startupIngress: expect.anything() })
)
expect(mux.request).not.toHaveBeenCalledWith('pty.spawn', expect.anything())
})
it('reattaches scoped app ids using raw relay ids', async () => {
mux.request.mockResolvedValue({ replay: 'buffered-output' })
@ -531,6 +566,28 @@ describe('SshPtyProvider', () => {
expect(handler).toHaveBeenCalledWith({ id: scopedPty1, data: 'output' })
})
it('forwards empty transformed relay spans without reinterpreting them', () => {
const handler = vi.fn()
provider.onData(handler)
const notifHandler = mux.onNotification.mock.calls[0][0]
notifHandler('pty.data', {
id: 'pty-1',
data: '',
rawLength: 9,
seq: 9,
transformed: true
})
expect(handler).toHaveBeenCalledWith({
id: scopedPty1,
data: '',
sequenceChars: 9,
seq: 9,
transformed: true
})
})
it('forwards pty.replay notifications to replay listeners', () => {
const handler = vi.fn()
provider.onReplay(handler)

View File

@ -2,8 +2,15 @@ import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
import type { IPtyProvider, PtyProcessInfo, PtySpawnOptions, PtySpawnResult } from './types'
import { toAppSshPtyId, toRelaySshPtyId } from './ssh-pty-id'
import { seedPowerlevel10kWizardEnv } from '../pty/powerlevel10k-wizard-env'
import { PTY_STARTUP_INGRESS_VERSION } from '../../shared/pty-startup-ingress'
type DataCallback = (payload: { id: string; data: string }) => void
type DataCallback = (payload: {
id: string
data: string
sequenceChars?: number
transformed?: boolean
seq?: number
}) => void
type ReplayCallback = (payload: { id: string; data: string }) => void
type ExitCallback = (payload: { id: string; code: number }) => void
type RemoteCliBridgeEnv = {
@ -56,7 +63,15 @@ export class SshPtyProvider implements IPtyProvider {
switch (method) {
case 'pty.data':
for (const cb of this.dataListeners) {
cb({ id: this.toAppPtyId(params.id as string), data: params.data as string })
cb({
id: this.toAppPtyId(params.id as string),
data: params.data as string,
...(typeof params.rawLength === 'number'
? { sequenceChars: params.rawLength as number }
: {}),
...(params.transformed === true ? { transformed: true } : {}),
...(typeof params.seq === 'number' ? { seq: params.seq as number } : {})
})
}
break
@ -167,7 +182,13 @@ export class SshPtyProvider implements IPtyProvider {
// remote hooks are disabled, but the relay still needs attach identity
// metadata to reject cross-generation PTY id collisions.
...(opts.paneKey ? { paneKey: opts.paneKey } : {}),
...(opts.tabId ? { tabId: opts.tabId } : {})
...(opts.tabId ? { tabId: opts.tabId } : {}),
...(opts.startupIngress
? {
startupIngressVersion: PTY_STARTUP_INGRESS_VERSION,
startupIngress: opts.startupIngress
}
: {})
})
return {
...(result as PtySpawnResult),
@ -261,6 +282,13 @@ export class SshPtyProvider implements IPtyProvider {
await this.mux.request('pty.clearBuffer', { id: this.toRelayPtyId(id) })
}
async closeStartupQueryAuthority(id: string): Promise<number> {
const result = (await this.mux.request('pty.closeStartupQueryAuthority', {
id: this.toRelayPtyId(id)
})) as { appliedSeq?: number }
return result.appliedSeq ?? 0
}
acknowledgeDataEvent(id: string, charCount: number): void {
this.mux.notify('pty.ackData', { id: this.toRelayPtyId(id), charCount })
}

View File

@ -18,29 +18,23 @@ import type {
SearchResult
} from '../../shared/types'
import type { GitHistoryOptions, GitHistoryResult } from '../../shared/git-history'
import type { PtyStartupIngressIntent } from '../../shared/pty-startup-ingress'
import type { CommitMessageDraftContext } from '../../shared/commit-message-generation'
import type { WorkspaceSpaceDirectoryScanResult } from '../../shared/workspace-space-types'
import type { StartupCommandDelivery } from '../../shared/codex-startup-delivery'
import type { TerminalOscLinkRange } from '../../shared/terminal-osc-link-ranges'
import type { TerminalGitHubPRLink } from '../../shared/terminal-github-pr-link-detector'
import type { GitProviderStatusOptions } from './git-provider-status-options'
import type { PtyBackgroundStreamEvent, PtyDataEvent } from './pty-provider-events'
import type { PtySpawnResult } from './pty-spawn-result'
export type {
PtyBackgroundStreamEvent,
PtyDataEvent,
PtyTransientFact
} from './pty-provider-events'
// ─── PTY Provider ───────────────────────────────────────────────────
/** Notification-bearing fact a thinning transport detected while it held
* scan authority for a backgrounded PTY (see onBackgroundStreamEvent). */
export type PtyTransientFact =
| { kind: 'bell' }
| { kind: 'command-finished'; exitCode: number | null }
| { kind: 'pr-link'; link: TerminalGitHubPRLink }
| { kind: '2031-subscribe' }
export type PtyBackgroundStreamEvent =
| { id: string; kind: 'backgroundMarker'; background: boolean; scanSeedAnsi?: string }
| { id: string; kind: 'dataGap'; droppedChars: number; sequenceChars?: number }
| { id: string; kind: 'transientFact'; fact: PtyTransientFact }
export type PtyProviderBufferSnapshot = {
data: string
/** Authoritative normal buffer captured beside an alternate-screen frame. */
@ -98,6 +92,8 @@ export type PtySpawnOptions = {
* through spawn options keeps local PTY and daemon PTY semantics aligned
* without promoting pwsh into a separate shell family. */
terminalWindowsPowerShellImplementation?: 'auto' | 'powershell.exe' | 'pwsh.exe'
/** Fresh-spawn-only source authority installed before any PTY output is released. */
startupIngress?: PtyStartupIngressIntent
}
export type { PtySpawnResult }
@ -170,6 +166,8 @@ export type IPtyProvider = {
getCwd(id: string): Promise<string>
getInitialCwd(id: string): Promise<string>
clearBuffer(id: string): Promise<void>
/** Ordered handoff from startup source authority to the live/hidden view authority. */
closeStartupQueryAuthority?: (id: string) => Promise<number> | number
acknowledgeDataEvent(id: string, charCount: number): void
hasChildProcesses(id: string): Promise<boolean>
getForegroundProcess(id: string): Promise<string | null>
@ -180,9 +178,7 @@ export type IPtyProvider = {
listProcesses(): Promise<PtyProcessInfo[]>
getDefaultShell(): Promise<string>
getProfiles(): Promise<{ name: string; path: string }[]>
onData(
callback: (payload: { id: string; data: string; sequenceChars?: number }) => void
): () => void
onData(callback: (payload: PtyDataEvent) => void): () => void
onReplay(callback: (payload: { id: string; data: string }) => void): () => void
onExit(callback: (payload: { id: string; code: number }) => void): () => void
}

View File

@ -2282,7 +2282,12 @@ export class OrcaRuntimeService {
// without polling. Keyed by ptyId for O(1) lookup per data event.
private dataListeners = new Map<
string,
Set<(data: string, meta?: { seq?: number; rawLength?: number; cwd?: string }) => void>
Set<
(
data: string,
meta?: { seq?: number; rawLength?: number; transformed?: boolean; cwd?: string }
) => void
>
>()
// Why: startup draft paste can subscribe after the agent already emitted its
// ready marker. Keep a bounded raw buffer so fast startup output is replayed.
@ -6064,7 +6069,13 @@ export class OrcaRuntimeService {
* Handles incoming data from a PTY process, running agent detection,
* updating terminal tail buffers, and triggering foreground agent refreshes.
*/
onPtyData(ptyId: string, data: string, at: number, sequenceChars = data.length): number {
onPtyData(
ptyId: string,
data: string,
at: number,
sequenceChars = data.length,
transformed = false
): number {
const outputSequence = (this.ptyOutputSequenceById.get(ptyId) ?? 0) + sequenceChars
this.ptyOutputSequenceById.set(ptyId, outputSequence)
this.providerModeTrackersByPtyId.get(ptyId)?.scan(data)
@ -6282,7 +6293,8 @@ export class OrcaRuntimeService {
if (listeners) {
const meta = {
seq: outputSequence,
rawLength: data.length,
rawLength: sequenceChars,
...(transformed ? { transformed: true } : {}),
...(cwdChanged && cwd !== null ? { cwd } : {})
}
for (const listener of listeners) {
@ -7133,7 +7145,10 @@ export class OrcaRuntimeService {
subscribeToTerminalData(
ptyId: string,
listener: (data: string, meta?: { seq?: number; rawLength?: number; cwd?: string }) => void
listener: (
data: string,
meta?: { seq?: number; rawLength?: number; transformed?: boolean; cwd?: string }
) => void
): () => void {
return addListenerToMap(this.dataListeners, ptyId, listener)
}

View File

@ -140,11 +140,17 @@ type TerminalOutputChunk = {
meta?: TerminalOutputMeta
}
type TerminalOutputMeta = { seq?: number; rawLength?: number; cwd?: string }
type TerminalOutputMeta = {
seq?: number
rawLength?: number
transformed?: boolean
cwd?: string
}
type TerminalOutputFrameChunk = {
bytes: Uint8Array<ArrayBufferLike>
seq?: number
opcode?: TerminalStreamOpcode
}
function createTerminalOutputBatcher(onFlush: (data: string, meta?: TerminalOutputMeta) => void): {
@ -156,6 +162,7 @@ function createTerminalOutputBatcher(onFlush: (data: string, meta?: TerminalOutp
let bytes = 0
let lastSeq: number | undefined
let pendingCwd: string | undefined
let pendingRawLength = 0
let timer: ReturnType<typeof setTimeout> | null = null
const clearTimer = (): void => {
@ -168,14 +175,14 @@ function createTerminalOutputBatcher(onFlush: (data: string, meta?: TerminalOutp
const flush = (): void => {
clearTimer()
if (chunks.length === 0) {
if (chunks.length === 0 && pendingRawLength === 0) {
return
}
const data = chunks.length === 1 ? chunks[0]! : chunks.join('')
const meta =
typeof lastSeq === 'number' || pendingCwd !== undefined
? {
...(typeof lastSeq === 'number' ? { seq: lastSeq, rawLength: data.length } : {}),
...(typeof lastSeq === 'number' ? { seq: lastSeq, rawLength: pendingRawLength } : {}),
...(pendingCwd !== undefined ? { cwd: pendingCwd } : {})
}
: undefined
@ -183,12 +190,19 @@ function createTerminalOutputBatcher(onFlush: (data: string, meta?: TerminalOutp
bytes = 0
lastSeq = undefined
pendingCwd = undefined
pendingRawLength = 0
onFlush(data, meta)
}
return {
push(data: string, meta?: TerminalOutputMeta): void {
if (!data) {
const rawLength = meta?.rawLength ?? data.length
if (!data && rawLength === 0) {
return
}
if (meta?.transformed || rawLength !== data.length) {
flush()
onFlush(data, { ...meta, rawLength, transformed: true })
return
}
if (meta?.cwd !== undefined) {
@ -196,6 +210,7 @@ function createTerminalOutputBatcher(onFlush: (data: string, meta?: TerminalOutp
pendingCwd = meta.cwd
}
chunks.push(data)
pendingRawLength += rawLength
const remainingBudget = Math.max(1, TERMINAL_OUTPUT_BATCH_MAX_BYTES - bytes)
const measurement = measureTerminalStreamByteLength(data, {
stopAfterBytes: remainingBudget
@ -222,6 +237,7 @@ function createTerminalOutputBatcher(onFlush: (data: string, meta?: TerminalOutp
clearTimer()
chunks = []
bytes = 0
pendingRawLength = 0
}
}
}
@ -230,11 +246,19 @@ function* iterateTerminalOutputFrameChunks(
data: string,
meta?: TerminalOutputMeta
): Generator<TerminalOutputFrameChunk> {
const rawLength = meta?.rawLength ?? data.length
if (meta?.transformed || rawLength !== data.length) {
yield {
opcode: TerminalStreamOpcode.OutputSpan,
bytes: encodeTerminalStreamJson({ data, rawLength, transformed: true }),
seq: meta?.seq
}
return
}
if (!terminalStreamByteLengthExceeds(data, TERMINAL_STREAM_CHUNK_BYTES)) {
yield { bytes: encodeTerminalStreamText(data), seq: meta?.seq }
return
}
const rawLength = meta?.rawLength ?? data.length
const canPreserveChunkSeq = typeof meta?.seq === 'number' && rawLength === data.length
const shouldDelayFinalSeq = !canPreserveChunkSeq && typeof meta?.seq === 'number'
const startSeq = canPreserveChunkSeq ? meta.seq! - rawLength : undefined
@ -1640,7 +1664,12 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
stream: TerminalMultiplexStream,
chunk: TerminalOutputFrameChunk
): void => {
sendFrame(stream.streamId, TerminalStreamOpcode.Output, chunk.bytes, chunk.seq)
sendFrame(
stream.streamId,
chunk.opcode ?? TerminalStreamOpcode.Output,
chunk.bytes,
chunk.seq
)
if (stream.ackOutput) {
stream.ackInFlightBytes += chunk.bytes.byteLength
ackTotalInFlightBytes += chunk.bytes.byteLength
@ -2802,7 +2831,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
)
}
for (const chunk of iterateTerminalOutputFrameChunks(data, meta)) {
sendFrame(TerminalStreamOpcode.Output, chunk.bytes, chunk.seq)
sendFrame(chunk.opcode ?? TerminalStreamOpcode.Output, chunk.bytes, chunk.seq)
}
})
unregisterBinaryHandler =
@ -3199,6 +3228,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
if (!initialOutputOverflowed) {
for (const item of bufferedOutput) {
let uncoveredData = getOutputAfterSnapshotSeq(item, snapshotOutputSeq)
let uncoveredMeta = item.meta
if (
uncoveredData &&
uncoveredData !== item.data &&
@ -3206,6 +3236,9 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
typeof item.meta?.seq === 'number' &&
typeof item.meta.rawLength === 'number'
) {
if (item.meta.rawLength === item.data.length) {
uncoveredMeta = { ...item.meta, rawLength: uncoveredData.length }
}
uncoveredData = stripSnapshotBoundaryQuerySuffixes(
uncoveredData,
snapshotOutputSeq,
@ -3214,7 +3247,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
)
}
if (uncoveredData) {
outputBatcher.push(uncoveredData, item.meta)
outputBatcher.push(uncoveredData, uncoveredMeta)
}
}
}

View File

@ -0,0 +1,112 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { SshRelaySession } from './ssh-relay-session'
import { createMockDeps, mockDeploySuccess } from './ssh-relay-session-test-fixtures'
const { muxRequestMock } = vi.hoisted(() => ({ muxRequestMock: vi.fn() }))
vi.mock('./ssh-relay-deploy', () => ({ deployAndLaunchRelay: vi.fn() }))
vi.mock('./ssh-channel-multiplexer', () => ({
SshChannelMultiplexer: class MockSshChannelMultiplexer {
notify = vi.fn()
request = muxRequestMock
onNotification = vi.fn().mockReturnValue(() => {})
onRequest = vi.fn().mockReturnValue(() => {})
onDispose = vi.fn().mockReturnValue(() => {})
dispose = vi.fn()
isDisposed = vi.fn().mockReturnValue(false)
}
}))
vi.mock('../agent-hooks/remote-managed-hook-installers', () => ({
installRemoteManagedAgentHooks: vi.fn().mockResolvedValue([])
}))
vi.mock('../providers/ssh-pty-provider', () => ({
isSshPtyNotFoundError: vi.fn().mockReturnValue(false),
isSshPtyIdentityMismatchError: vi.fn().mockReturnValue(false),
SshPtyProvider: class MockSshPtyProvider {
onData = vi.fn().mockReturnValue(() => {})
onReplay = vi.fn().mockReturnValue(() => {})
onExit = vi.fn().mockReturnValue(() => {})
attach = vi.fn().mockResolvedValue(undefined)
attachForReconnect = vi.fn().mockResolvedValue({})
dispose = vi.fn()
}
}))
vi.mock('../providers/ssh-filesystem-provider', () => ({
SshFilesystemProvider: class MockSshFilesystemProvider {
dispose = vi.fn()
}
}))
vi.mock('../providers/ssh-git-provider', () => ({
SshGitProvider: class MockSshGitProvider {}
}))
vi.mock('../ipc/pty', () => ({
registerSshPtyProvider: vi.fn(),
unregisterSshPtyProvider: vi.fn(),
getSshPtyProvider: vi.fn().mockReturnValue({ dispose: vi.fn() }),
getPtyIdsForConnection: vi.fn().mockReturnValue([]),
clearPtyOwnershipForConnection: vi.fn(),
clearProviderPtyState: vi.fn(),
deletePtyOwnership: vi.fn(),
setPtyOwnership: vi.fn()
}))
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
registerSshFilesystemProvider: vi.fn(),
unregisterSshFilesystemProvider: vi.fn(),
getSshFilesystemProvider: vi.fn().mockReturnValue({ dispose: vi.fn() })
}))
vi.mock('../providers/ssh-git-dispatch', () => ({
registerSshGitProvider: vi.fn(),
unregisterSshGitProvider: vi.fn()
}))
const { registerSshPtyProvider } = await import('../ipc/pty')
describe('SshRelaySession data delivery', () => {
beforeEach(() => {
vi.clearAllMocks()
muxRequestMock.mockResolvedValue([])
mockDeploySuccess()
})
it('delivers empty transformed relay spans with raw sequence metadata', async () => {
const { mockConn, mockStore, mockPortForward, getMainWindow, mockWindow } = createMockDeps()
const runtime = { onPtyData: vi.fn(() => 17), onPtyExit: vi.fn() }
const session = new SshRelaySession(
'target-1',
getMainWindow,
mockStore,
mockPortForward,
runtime as never
)
await session.establish(mockConn)
const ptyProvider = vi.mocked(registerSshPtyProvider).mock.calls[0]?.[1] as unknown as {
onData: ReturnType<typeof vi.fn>
}
const onData = ptyProvider.onData.mock.calls[0]?.[0] as (payload: {
id: string
data: string
sequenceChars?: number
transformed?: boolean
}) => void
onData({ id: 'ssh-pty-1', data: '', sequenceChars: 9, transformed: true })
expect(runtime.onPtyData).toHaveBeenCalledWith('ssh-pty-1', '', expect.any(Number), 9, true)
expect(mockWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
id: 'ssh-pty-1',
data: '',
sequenceChars: 9,
transformed: true,
seq: 17,
rawLength: 9
})
})
})

View File

@ -65,8 +65,7 @@ vi.mock('../ipc/pty', () => ({
clearPtyOwnershipForConnection: vi.fn(),
clearProviderPtyState: vi.fn(),
deletePtyOwnership: vi.fn(),
setPtyOwnership: vi.fn(),
answerStartupTerminalColorQueriesForPty: vi.fn((_id: string, data: string) => data)
setPtyOwnership: vi.fn()
}))
vi.mock('../providers/ssh-filesystem-dispatch', () => ({

View File

@ -73,8 +73,7 @@ vi.mock('../ipc/pty', () => ({
clearPtyOwnershipForConnection: vi.fn(),
clearProviderPtyState: vi.fn(),
deletePtyOwnership: vi.fn(),
setPtyOwnership: vi.fn(),
answerStartupTerminalColorQueriesForPty: vi.fn((_id: string, data: string) => data)
setPtyOwnership: vi.fn()
}))
vi.mock('../providers/ssh-filesystem-dispatch', () => ({
@ -152,7 +151,9 @@ describe('SshRelaySession', () => {
expect(runtime.onPtyData).toHaveBeenCalledWith(
'ssh-pty-1',
'hidden ssh output',
expect.any(Number)
expect.any(Number),
'hidden ssh output'.length,
undefined
)
expect(mockWindow.webContents.send).toHaveBeenCalledTimes(1)
// Why out-of-band: an in-band empty pty:data sentinel is ambiguous with
@ -207,7 +208,8 @@ describe('SshRelaySession', () => {
expect(mockWindow.webContents.send).toHaveBeenCalledWith('pty:data', {
id: 'ssh-pty-1',
data: 'still delivered'
data: 'still delivered',
rawLength: 'still delivered'.length
})
})

View File

@ -41,8 +41,7 @@ import {
clearPtyOwnershipForConnection,
clearProviderPtyState,
deletePtyOwnership,
setPtyOwnership,
answerStartupTerminalColorQueriesForPty
setPtyOwnership
} from '../ipc/pty'
import {
recordHiddenRendererPtyDataDrop,
@ -1120,8 +1119,14 @@ export class SshRelaySession {
private wireUpPtyEvents(ptyProvider: SshPtyProvider): void {
ptyProvider.onData((payload) => {
const seq = this.runtime?.onPtyData(payload.id, payload.data, Date.now())
const rendererData = answerStartupTerminalColorQueriesForPty(payload.id, payload.data)
const rawLength = payload.sequenceChars ?? payload.data.length
const seq = this.runtime?.onPtyData(
payload.id,
payload.data,
Date.now(),
rawLength,
payload.transformed
)
const win = this.getMainWindow()
if (!win || win.isDestroyed()) {
return
@ -1133,7 +1138,7 @@ export class SshRelaySession {
// OSC-9999-only chunks legitimately strip to empty in the renderer.
const store = this.store as { getSettings?: Store['getSettings'] }
if (shouldDropHiddenRendererPtyData(payload.id, store.getSettings?.())) {
const drop = recordHiddenRendererPtyDataDrop(payload.id, payload.data.length)
const drop = recordHiddenRendererPtyDataDrop(payload.id, rawLength)
if (drop.shouldEmitRestoreMarker) {
win.webContents.send('pty:modelRestoreNeeded', {
id: payload.id,
@ -1143,16 +1148,12 @@ export class SshRelaySession {
}
return
}
// Why: startup color-query answering can strip query-only chunks to
// empty; skip empty sends and only attach seq metadata when the chunk
// reaches the renderer unmodified (seq tracks raw stream offsets).
if (rendererData.length > 0) {
if (payload.data.length > 0 || payload.transformed) {
win.webContents.send('pty:data', {
...payload,
data: rendererData,
...(rendererData === payload.data && typeof seq === 'number'
? { seq, rawLength: payload.data.length }
: {})
...(typeof seq === 'number' ? { seq } : {}),
rawLength,
...(payload.transformed ? { transformed: true } : {})
})
}
})

View File

@ -1439,6 +1439,7 @@ export type PreloadApi = {
data: string
seq?: number
rawLength?: number
transformed?: boolean
background?: boolean
droppedOutput?: boolean
}) => void

View File

@ -1061,6 +1061,7 @@ const api = {
data: string
seq?: number
rawLength?: number
transformed?: boolean
background?: boolean
droppedOutput?: boolean
}) => void
@ -1072,6 +1073,7 @@ const api = {
data: string
seq?: number
rawLength?: number
transformed?: boolean
background?: boolean
droppedOutput?: boolean
}

View File

@ -10,6 +10,7 @@ import {
resolveSetupAgentSequenceLaunchCommand,
SETUP_AGENT_SEQUENCE_STARTUP_COMMAND_ENV
} from '../shared/setup-agent-sequencing'
import { PTY_STARTUP_INGRESS_VERSION } from '../shared/pty-startup-ingress'
const { mockPtySpawn, mockPtyInstance } = vi.hoisted(() => ({
mockPtySpawn: vi.fn(),
@ -968,6 +969,72 @@ describe('PtyHandler', () => {
expect(dispatcher.notify).toHaveBeenCalledWith('pty.data', { id: 'pty-1', data: 'hello world' })
})
it('consumes capable startup queries before relay replay and fanout', async () => {
let dataCallback: ((data: string) => void) | undefined
const term = {
...mockPtyInstance,
onData: vi.fn((cb: (data: string) => void) => {
dataCallback = cb
}),
onExit: vi.fn()
}
mockPtySpawn.mockReturnValue(term)
await dispatcher.callRequest('pty.spawn', {
startupIngressVersion: PTY_STARTUP_INGRESS_VERSION,
startupIngress: {
colors: { foreground: '#2e3434', background: '#ffffff' },
deadlineMs: 5_000
}
})
const query = '\x1b]10;?\x07'
dataCallback!(query)
dataCallback!('prompt')
vi.advanceTimersByTime(8)
expect(term.write).toHaveBeenCalledWith('\x1b]10;rgb:2e2e/3434/3434\x1b\\')
expect(dispatcher.notify).toHaveBeenCalledWith('pty.data', {
id: 'pty-1',
data: '',
rawLength: query.length,
seq: query.length,
transformed: true
})
expect(dispatcher.notify).toHaveBeenCalledWith('pty.data', { id: 'pty-1', data: 'prompt' })
await expect(
dispatcher.callRequest('pty.attach', {
id: 'pty-1',
suppressReplayNotification: true
})
).resolves.toEqual({ replay: 'prompt' })
})
it('leaves startup queries untouched for an unsupported relay capability version', async () => {
let dataCallback: ((data: string) => void) | undefined
const term = {
...mockPtyInstance,
onData: vi.fn((cb: (data: string) => void) => {
dataCallback = cb
}),
onExit: vi.fn()
}
mockPtySpawn.mockReturnValue(term)
await dispatcher.callRequest('pty.spawn', {
startupIngressVersion: PTY_STARTUP_INGRESS_VERSION - 1,
startupIngress: {
colors: { foreground: '#2e3434', background: '#ffffff' },
deadlineMs: 5_000
}
})
const query = '\x1b]10;?\x07'
dataCallback!(query)
vi.advanceTimersByTime(8)
expect(term.write).not.toHaveBeenCalled()
expect(dispatcher.notify).toHaveBeenCalledWith('pty.data', { id: 'pty-1', data: query })
})
it('coalesces background PTY output before notifying the client', async () => {
let dataCallback: ((data: string) => void) | undefined
mockPtySpawn.mockReturnValue({

View File

@ -39,6 +39,12 @@ import {
} from '../shared/git-credential-prompt-env'
import { isTuiAgent } from '../shared/tui-agent-config'
import { forceKillPosixPtyProcessGroups } from '../main/pty/posix-pty-process-groups'
import {
PTY_STARTUP_INGRESS_VERSION,
PtyStartupIngress,
parsePtyStartupIngressIntent,
type PtyIngressEmission
} from '../shared/pty-startup-ingress'
function isMissingNodePtyNativeBinding(error: unknown): boolean {
return (
@ -78,10 +84,15 @@ type ManagedPty = {
physicalExit?: PhysicalExitTracker
forceKillSent?: boolean
gracefulKillSent?: boolean
startupIngress?: PtyStartupIngress
startupIngressIntent?: ReturnType<typeof parsePtyStartupIngressIntent>
}
type PendingPtyOutput = {
data: string
rawLength?: number
transformed?: boolean
seq?: number
}
type ManagedStartupCommand = {
@ -471,6 +482,9 @@ export class PtyHandler {
}
private appendReplayBuffer(managed: ManagedPty, data: string): void {
if (data.length === 0) {
return
}
managed.buffered += data
if (managed.buffered.length > REPLAY_BUFFER_MAX) {
managed.buffered = managed.buffered.slice(-REPLAY_BUFFER_MAX)
@ -504,8 +518,7 @@ export class PtyHandler {
if (startup.scanState) {
const heldBytes = drainShellReadyHeldBytes(startup.scanState)
if (heldBytes) {
this.appendReplayBuffer(managed, heldBytes)
this.enqueuePtyOutput(managed.id, heldBytes)
managed.startupIngress?.accept(heldBytes)
}
}
const submit = process.platform === 'win32' ? '\r' : '\n'
@ -525,6 +538,22 @@ export class PtyHandler {
private wireAndStore(managed: ManagedPty): void {
managed.physicalExit = new PhysicalExitTracker()
this.ptys.set(managed.id, managed)
const emitIngressData = (emission: PtyIngressEmission): void => {
const rawLength = emission.rawEndSeq - emission.rawStartSeq
this.appendReplayBuffer(managed, emission.data)
this.enqueuePtyOutput(
managed.id,
emission.data,
emission.transformed || rawLength !== emission.data.length
? { rawLength, seq: emission.rawEndSeq, transformed: true }
: {}
)
}
managed.startupIngress ??= new PtyStartupIngress({
...(managed.startupIngressIntent ? { intent: managed.startupIngressIntent } : {}),
write: (data) => managed.pty.write(data),
onEmission: emitIngressData
})
managed.pty.onData((data: string) => {
const startup = managed.startupCommand
if (startup?.waitForShellReady && startup.scanState && !startup.delivered) {
@ -534,8 +563,7 @@ export class PtyHandler {
this.scheduleStartupCommandDelivery(managed, STARTUP_COMMAND_WRITE_DELAY_MS)
}
}
this.appendReplayBuffer(managed, data)
this.enqueuePtyOutput(managed.id, data)
managed.startupIngress?.accept(data)
})
managed.pty.onExit(({ exitCode }: { exitCode: number }) => {
managed.physicalExit?.markExited()
@ -560,6 +588,7 @@ export class PtyHandler {
managed.killTimer = undefined
}
this.clearStartupCommandTimer(managed)
this.releaseRelayIngress(managed)
this.flushPtyOutput(managed.id)
this.dispatcher.notify('pty.exit', { id: managed.id, code: exitCode })
this.notifyExitListener(managed)
@ -572,6 +601,17 @@ export class PtyHandler {
})
}
private releaseRelayIngress(managed: ManagedPty): void {
const startupCommand = managed.startupCommand
const scanState = startupCommand?.scanState
if (scanState) {
const held = drainShellReadyHeldBytes(scanState)
startupCommand.scanState = null
managed.startupIngress?.accept(held)
}
managed.startupIngress?.drainAndClose()
}
private notifyExitListener(managed: ManagedPty): void {
if (managed.exitListenerNotified) {
return
@ -605,6 +645,9 @@ export class PtyHandler {
this.dispatcher.onRequest('pty.serialize', (p) => this.serialize(p))
this.dispatcher.onRequest('pty.revive', (p) => this.revive(p))
this.dispatcher.onRequest('pty.getProfiles', async () => listShellProfiles())
this.dispatcher.onRequest('pty.closeStartupQueryAuthority', (p) =>
this.closeStartupQueryAuthority(p)
)
this.dispatcher.onNotification('pty.data', (p) => this.writeData(p))
this.dispatcher.onNotification('pty.resize', (p) => this.resize(p))
@ -620,6 +663,17 @@ export class PtyHandler {
return data.length <= INTERACTIVE_REDRAW_MAX_CHARS && data.includes('\x1b[')
}
private async closeStartupQueryAuthority(
params: Record<string, unknown>
): Promise<{ appliedSeq: number }> {
const id = params.id as string
const managed = this.ptys.get(id)
if (!managed || managed.disposed) {
throw new Error(`PTY "${id}" not found`)
}
return { appliedSeq: managed.startupIngress?.closeQueryAuthority() ?? 0 }
}
private shouldSendInteractiveOutputNow(id: string, data: string): boolean {
const lastInputAt = this.lastInputAtByPty.get(id)
const now = performance.now()
@ -640,15 +694,35 @@ export class PtyHandler {
return true
}
private enqueuePtyOutput(id: string, data: string): void {
private enqueuePtyOutput(
id: string,
data: string,
meta: { rawLength?: number; transformed?: boolean; seq?: number } = {}
): void {
const existing = this.pendingOutputByPty.get(id)
const pending = { data: (existing?.data ?? '') + data }
if (meta.transformed === true) {
// Why: transformed spans have no raw-to-clean slice mapping, so neither
// side of their boundary may be folded into the relay's output batch.
if (existing) {
this.flushPtyOutput(id)
}
this.dispatcher.notify('pty.data', { id, data, ...meta })
return
}
const pending: PendingPtyOutput = { data: (existing?.data ?? '') + data }
if (existing?.rawLength !== undefined || meta.rawLength !== undefined) {
pending.rawLength =
(existing?.rawLength ?? existing?.data.length ?? 0) + (meta.rawLength ?? data.length)
}
if (meta.seq !== undefined) {
pending.seq = meta.seq
}
if (this.shouldSendInteractiveOutputNow(id, pending.data)) {
this.pendingOutputByPty.delete(id)
this.clearOutputFlushTimerIfIdle()
// Why: remote agent TUIs redraw around each keystroke. Background relay
// batching should reduce SSH chatter, not add visible input echo delay.
this.dispatcher.notify('pty.data', { id, data: pending.data })
this.dispatcher.notify('pty.data', { id, ...pending })
return
}
this.pendingOutputByPty.set(id, pending)
@ -670,12 +744,31 @@ export class PtyHandler {
break
}
this.pendingOutputByPty.delete(id)
const chunk = pending.data.slice(0, PTY_OUTPUT_FLUSH_CHUNK_CHARS)
const remaining = pending.data.slice(PTY_OUTPUT_FLUSH_CHUNK_CHARS)
const chunk = pending.transformed
? pending.data
: pending.data.slice(0, PTY_OUTPUT_FLUSH_CHUNK_CHARS)
const remaining = pending.transformed ? '' : pending.data.slice(PTY_OUTPUT_FLUSH_CHUNK_CHARS)
if (remaining) {
this.pendingOutputByPty.set(id, { data: remaining })
this.pendingOutputByPty.set(id, {
data: remaining,
...(pending.rawLength === undefined ? {} : { rawLength: remaining.length }),
seq: pending.seq
})
}
this.dispatcher.notify('pty.data', { id, data: chunk })
const chunkRawLength = pending.transformed
? pending.rawLength
: pending.rawLength === undefined
? undefined
: chunk.length
const chunkSeq =
pending.seq === undefined ? undefined : pending.seq - (pending.data.length - chunk.length)
this.dispatcher.notify('pty.data', {
id,
data: chunk,
...(chunkSeq === undefined ? {} : { seq: chunkSeq }),
...(chunkRawLength === undefined ? {} : { rawLength: chunkRawLength }),
...(pending.transformed ? { transformed: true } : {})
})
writes++
}
if (this.pendingOutputByPty.size > 0 && writes > 0) {
@ -691,7 +784,7 @@ export class PtyHandler {
return
}
this.pendingOutputByPty.delete(id)
this.dispatcher.notify('pty.data', { id, data: pending.data })
this.dispatcher.notify('pty.data', { id, ...pending })
this.clearOutputFlushTimerIfIdle()
}
@ -891,6 +984,12 @@ export class PtyHandler {
tabId: typeof params.tabId === 'string' ? params.tabId : tabId
}
const worktreeId = typeof env?.ORCA_WORKTREE_ID === 'string' ? env.ORCA_WORKTREE_ID : undefined
const startupIngressIntent =
params.startupIngressVersion === PTY_STARTUP_INGRESS_VERSION
? parsePtyStartupIngressIntent(params.startupIngress, {
allowWindowsEchoProjection: false
})
: undefined
const managed: ManagedPty = {
id,
pty: term,
@ -903,6 +1002,7 @@ export class PtyHandler {
...(explicitTerm !== undefined ? { explicitTerm } : {}),
envToDelete,
gitCredentialPromptGuarded,
...(startupIngressIntent ? { startupIngressIntent } : {}),
...(terminalHandle ? { terminalHandle } : {}),
...(shouldProviderDeliverCommand
? {
@ -958,6 +1058,8 @@ export class PtyHandler {
// grace window already takes.
if (managed.pty.pid && !isProcessAlive(managed.pty.pid)) {
managed.physicalExit?.markExited()
this.releaseRelayIngress(managed)
this.flushPtyOutput(id)
this.notifyExitListener(managed)
disposeManagedPty(managed)
this.ptys.delete(id)
@ -984,6 +1086,8 @@ export class PtyHandler {
throw new Error(`PTY "${id}" not found (identity mismatch)`)
}
managed.startupIngress?.snapshotBarrier()
// Replay buffered output. During pty.spawn({ sessionId }) the renderer has
// not registered replay handlers yet, so return the bytes to the caller
// instead of notifying them too early.
@ -1175,6 +1279,7 @@ export class PtyHandler {
const id = params.id as string
const managed = this.ptys.get(id)
if (managed && !managed.disposed) {
managed.startupIngress?.snapshotBarrier()
managed.pty.clear()
}
}
@ -1385,6 +1490,10 @@ export class PtyHandler {
private async disposePtys(waitForPhysicalExit: boolean): Promise<void> {
this.cancelGraceTimer()
await this.waitForPendingPtyCreations()
for (const managed of this.ptys.values()) {
this.releaseRelayIngress(managed)
this.flushPtyOutput(managed.id)
}
if (this.outputFlushTimer !== null) {
clearTimeout(this.outputFlushTimer)
this.outputFlushTimer = null
@ -1414,6 +1523,7 @@ export class PtyHandler {
managed.killTimer = undefined
}
this.clearStartupCommandTimer(managed)
this.releaseRelayIngress(managed)
// Why: relay exit must retain the native owner until SIGKILL is accepted
// (with one bounded retry) or onExit proves the process is already gone.
await this.requestForceKillForRelayShutdown(managed)

View File

@ -7085,6 +7085,7 @@ export function connectPanePty(
return
}
if (!foreground && orderedRendererData.length === 0) {
recordRendererOrderedSeq(rendererMeta)
schedulePendingStartupCommandDelivery()
return
}

View File

@ -37,6 +37,7 @@ import { installTerminalFreezeReport } from './terminal-freeze-report'
export type PtyDataMeta = {
seq?: number
rawLength?: number
transformed?: boolean
background?: boolean
/** Main dropped this PTY's buffered output at the pending cap; the pane
* must repaint from the main-owned snapshot instead of the live stream. */
@ -170,6 +171,7 @@ function handleDispatchedPtyData(payload: {
data: string
seq?: number
rawLength?: number
transformed?: boolean
background?: boolean
droppedOutput?: boolean
}): void {
@ -182,6 +184,10 @@ function handleDispatchedPtyData(payload: {
meta ??= {}
meta.rawLength = payload.rawLength
}
if (payload.transformed === true) {
meta ??= {}
meta.transformed = true
}
if (payload.background === true) {
meta ??= {}
meta.background = true

View File

@ -105,6 +105,23 @@ class FakeMultiplexServer {
this.send(TerminalStreamOpcode.Output, encodeTerminalStreamText(text), this.cursorUnits)
}
outputSpan(data: string, rawLength: number): void {
this.cursorUnits += rawLength
this.send(
TerminalStreamOpcode.OutputSpan,
encodeTerminalStreamJson({ data, rawLength, transformed: true }),
this.cursorUnits
)
}
malformedOutputSpan(): void {
this.send(
TerminalStreamOpcode.OutputSpan,
encodeTerminalStreamJson({ data: 'framing must not render' }),
this.cursorUnits
)
}
flushHeldManualSnapshot(): void {
if (this.heldManualRequestId === null) {
throw new Error('No manual snapshot is held')
@ -144,24 +161,29 @@ describe('remote terminal frame-drop resync', () => {
async function subscribeClient(): Promise<{
data: string[]
metas: { seq?: number; rawLength?: number; transformed?: boolean }[]
snapshots: string[]
stream: RemoteRuntimeMultiplexedTerminal
}> {
const data: string[] = []
const metas: { seq?: number; rawLength?: number; transformed?: boolean }[] = []
const snapshots: string[] = []
const multiplexer = getRemoteRuntimeTerminalMultiplexer('env-1')
const stream = await multiplexer.subscribeTerminal({
terminal: 'terminal-1',
client: { id: 'desktop-1', type: 'desktop' },
callbacks: {
onData: (chunk) => data.push(chunk),
onData: (chunk, meta) => {
data.push(chunk)
metas.push(meta ?? {})
},
onSnapshot: (chunk) => snapshots.push(chunk)
}
})
// Let the initial snapshot round-trip settle.
await Promise.resolve()
await Promise.resolve()
return { data, snapshots, stream }
return { data, metas, snapshots, stream }
}
it('detects a dropped Output frame via the seq gap and resyncs', async () => {
@ -197,6 +219,27 @@ describe('remote terminal frame-drop resync', () => {
expect(snapshots).toEqual(['INITIAL'])
})
it('delivers an empty transformed span with its raw sequence metadata', async () => {
const { data, metas, snapshots } = await subscribeClient()
server.outputSpan('', 9)
expect(data).toEqual([''])
expect(metas).toEqual([{ seq: 9, rawLength: 9, transformed: true }])
expect(snapshots).toEqual(['INITIAL'])
})
it('requests an authoritative resync instead of rendering malformed span JSON', async () => {
const { data, snapshots } = await subscribeClient()
server.malformedOutputSpan()
await Promise.resolve()
await Promise.resolve()
expect(data).toEqual([])
expect(snapshots).toEqual(['INITIAL', '\x1b[2J\x1b[3J\x1b[HRECOVERED'])
})
it('uses UTF-16 sequence units when detecting gaps in multibyte output', async () => {
const { data, snapshots } = await subscribeClient()

View File

@ -37,7 +37,7 @@ type TerminalMultiplexEvent =
| { type: string; streamId?: number; [key: string]: unknown }
export type RemoteRuntimeMultiplexedTerminalCallbacks = {
onData: (data: string, meta?: { seq?: number; rawLength?: number }) => void
onData: (data: string, meta?: { seq?: number; rawLength?: number; transformed?: boolean }) => void
onSnapshot: (data: string, meta?: { pendingEscapeTailAnsi?: string }) => void
onSubscribed?: () => void
onEnd?: () => void
@ -437,10 +437,42 @@ class RemoteRuntimeTerminalMultiplexer {
if (!stream) {
return
}
if (frame.opcode === TerminalStreamOpcode.Output) {
const data = decodeTerminalStreamText(frame.payload)
if (
frame.opcode === TerminalStreamOpcode.Output ||
frame.opcode === TerminalStreamOpcode.OutputSpan
) {
const span =
frame.opcode === TerminalStreamOpcode.OutputSpan
? decodeTerminalStreamJson<{
data?: unknown
rawLength?: unknown
transformed?: unknown
}>(frame.payload)
: null
const validSpan =
frame.opcode !== TerminalStreamOpcode.OutputSpan ||
(typeof span?.data === 'string' &&
typeof span.rawLength === 'number' &&
Number.isSafeInteger(span.rawLength) &&
span.rawLength >= 0 &&
span.transformed === true)
const data =
frame.opcode === TerminalStreamOpcode.OutputSpan
? validSpan
? (span!.data as string)
: ''
: decodeTerminalStreamText(frame.payload)
try {
const rawLength = data.length
if (!validSpan) {
// Why: rendering malformed span JSON would expose protocol framing
// as terminal text and lose its raw sequence accounting.
this.requestResyncSnapshot(stream)
return
}
const rawLength =
frame.opcode === TerminalStreamOpcode.OutputSpan && typeof span?.rawLength === 'number'
? span.rawLength
: data.length
// Why: a resync snapshot is authoritative; discard live output while
// it is in flight, but still return transport credit in finally.
if (stream.resyncInFlight) {
@ -454,7 +486,11 @@ class RemoteRuntimeTerminalMultiplexer {
if (typeof seq === 'number') {
stream.expectedSeq = seq
}
stream.callbacks.onData(data, { seq, rawLength })
stream.callbacks.onData(data, {
seq,
rawLength,
...(frame.opcode === TerminalStreamOpcode.OutputSpan ? { transformed: true } : {})
})
} finally {
if (stream.acknowledgeOutput) {
if (shouldHoldE2eRemoteTerminalAck(stream.terminal)) {

View File

@ -0,0 +1,52 @@
import {
terminalOscColorQueryReplies,
type TerminalOscColorQueryReplyColors
} from './terminal-osc-color-reply'
export type PtyStartupIngressIntent = {
colors: TerminalOscColorQueryReplyColors
deadlineMs: number
echoProjection?: 'windows-conpty-esc-stripped'
}
export const PTY_STARTUP_INGRESS_VERSION = 1
export function parsePtyStartupIngressIntent(
value: unknown,
options: { allowWindowsEchoProjection: boolean }
): PtyStartupIngressIntent | undefined {
if (!value || typeof value !== 'object') {
return undefined
}
const record = value as Record<string, unknown>
const colors = record.colors
if (!colors || typeof colors !== 'object') {
return undefined
}
const colorRecord = colors as Record<string, unknown>
const normalizedColors = {
...(typeof colorRecord.foreground === 'string' ? { foreground: colorRecord.foreground } : {}),
...(typeof colorRecord.background === 'string' ? { background: colorRecord.background } : {})
}
if (
!terminalOscColorQueryReplies(normalizedColors, [10, 11]) ||
typeof record.deadlineMs !== 'number' ||
!Number.isFinite(record.deadlineMs) ||
record.deadlineMs < 0 ||
record.deadlineMs > 30_000
) {
return undefined
}
const projection = record.echoProjection
if (
projection !== undefined &&
(projection !== 'windows-conpty-esc-stripped' || !options.allowWindowsEchoProjection)
) {
return undefined
}
return {
colors: normalizedColors,
deadlineMs: record.deadlineMs,
...(projection === 'windows-conpty-esc-stripped' ? { echoProjection: projection } : {})
}
}

View File

@ -0,0 +1,158 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import {
PtyStartupIngress,
parsePtyStartupIngressIntent,
type PtyIngressEmission
} from './pty-startup-ingress'
const COLORS = { foreground: '#2e3434', background: '#ffffff' }
function createHarness(options: { projection?: boolean; nested?: (data: string) => void } = {}) {
const emissions: PtyIngressEmission[] = []
let ingress!: PtyStartupIngress
const writes: string[] = []
ingress = new PtyStartupIngress({
intent: {
colors: COLORS,
deadlineMs: 5_000,
...(options.projection ? { echoProjection: 'windows-conpty-esc-stripped' as const } : {})
},
write: (data) => {
writes.push(data)
options.nested?.(data)
},
onEmission: (emission) => emissions.push(emission)
})
return { ingress, writes, emissions }
}
function visible(emissions: readonly PtyIngressEmission[]): string {
return emissions.map((emission) => emission.data).join('')
}
describe('PtyStartupIngress', () => {
afterEach(() => vi.useRealTimers())
it('validates intent bounds and rejects a Windows projection on isolated hosts', () => {
const intent = {
colors: COLORS,
deadlineMs: 5_000,
echoProjection: 'windows-conpty-esc-stripped'
}
expect(parsePtyStartupIngressIntent(intent, { allowWindowsEchoProjection: true })).toEqual(
intent
)
expect(parsePtyStartupIngressIntent(intent, { allowWindowsEchoProjection: false })).toBe(
undefined
)
expect(
parsePtyStartupIngressIntent(
{ ...intent, deadlineMs: 30_001 },
{
allowWindowsEchoProjection: true
}
)
).toBeUndefined()
})
it('recognizes BEL/ST queries at every split and emits canonical replies', () => {
const query = '\x1b]10;?\x07\x1b]11;?\x1b\\'
for (let split = 0; split <= query.length; split += 1) {
const { ingress, writes, emissions } = createHarness()
ingress.accept(query.slice(0, split))
ingress.accept(query.slice(split))
ingress.drainAndClose()
expect(visible(emissions), `split ${split}`).toBe('')
expect(writes, `split ${split}`).toEqual([
'\x1b]10;rgb:2e2e/3434/3434\x1b\\',
'\x1b]11;rgb:ffff/ffff/ffff\x1b\\'
])
expect(emissions.reduce((sum, item) => sum + item.rawEndSeq - item.rawStartSeq, 0)).toBe(
query.length
)
}
})
it('suppresses the first echo immediately and keeps a later exact collision', () => {
const { ingress, emissions } = createHarness({ projection: true })
ingress.accept('\x1b]10;?\x07')
const projected = ']10;rgb:2e2e/3434/3434\\'
ingress.accept(projected)
ingress.accept(projected)
ingress.drainAndClose()
expect(visible(emissions)).toBe(projected)
})
it('matches each echo across every split without skipping an earlier FIFO candidate', () => {
const foregroundEcho = ']10;rgb:2e2e/3434/3434\\'
const backgroundEcho = ']11;rgb:ffff/ffff/ffff\\'
for (const projected of [foregroundEcho, backgroundEcho]) {
for (let split = 0; split <= projected.length; split += 1) {
const { ingress, emissions } = createHarness({ projection: true })
ingress.accept(projected === foregroundEcho ? '\x1b]10;?\x07' : '\x1b]11;?\x1b\\')
ingress.accept(projected.slice(0, split))
ingress.accept(projected.slice(split))
ingress.drainAndClose()
expect(visible(emissions), `${projected.slice(0, 3)} split ${split}`).toBe('')
}
}
const fifo = createHarness({ projection: true })
fifo.ingress.accept('\x1b]10;?;?\x1b\\')
fifo.ingress.accept(backgroundEcho)
fifo.ingress.accept(backgroundEcho)
fifo.ingress.drainAndClose()
expect(visible(fifo.emissions)).toBe(backgroundEcho)
})
it('releases partial echo bytes on mismatch, timeout, and snapshot barrier', () => {
vi.useFakeTimers()
const mismatch = createHarness({ projection: true })
mismatch.ingress.accept('\x1b]10;?\x07')
mismatch.ingress.accept(']10;rgb:2e2e/nope')
expect(visible(mismatch.emissions)).toBe(']10;rgb:2e2e/nope')
const timeout = createHarness({ projection: true })
timeout.ingress.accept('\x1b]10;?\x07')
timeout.ingress.accept(']10;rgb:2e2e/')
vi.advanceTimersByTime(5_000)
expect(visible(timeout.emissions)).toBe(']10;rgb:2e2e/')
const snapshot = createHarness({ projection: true })
snapshot.ingress.accept('\x1b]10;?\x07')
snapshot.ingress.accept(']10;rgb:2e2e/')
snapshot.ingress.snapshotBarrier()
expect(visible(snapshot.emissions)).toBe(']10;rgb:2e2e/')
snapshot.ingress.accept('\x1b]11;?\x07')
expect(snapshot.writes.at(-1)).toBe('\x1b]11;rgb:ffff/ffff/ffff\x1b\\')
})
it('serializes a synchronous nested provider callback after the consumed query span', () => {
const emissions: PtyIngressEmission[] = []
let ingress!: PtyStartupIngress
ingress = new PtyStartupIngress({
intent: { colors: COLORS, deadlineMs: 5_000 },
write: () => ingress.accept('nested'),
onEmission: (emission) => emissions.push(emission)
})
ingress.accept('before\x1b]10;?\x07after')
ingress.drainAndClose()
expect(emissions.map(({ data, transformed }) => ({ data, transformed }))).toEqual([
{ data: 'before', transformed: false },
{ data: '', transformed: true },
{ data: 'after', transformed: false },
{ data: 'nested', transformed: false }
])
})
it('ignores callbacks after teardown without recreating the raw sequence domain', () => {
const { ingress, emissions } = createHarness({ projection: true })
ingress.accept('\x1b]10;?\x07')
ingress.accept(']10;rgb:2e2e/')
const closedAt = ingress.drainAndClose()
ingress.accept('late')
expect(ingress.acceptedRawSequence).toBe(closedAt)
expect(visible(emissions)).toBe(']10;rgb:2e2e/')
})
})

View File

@ -0,0 +1,340 @@
import {
parseTerminalOscColorQuery,
terminalOscColorQueryReplies,
type TerminalOscColorQuerySlot
} from './terminal-osc-color-reply'
import type { PtyStartupIngressIntent } from './pty-startup-ingress-intent'
export {
PTY_STARTUP_INGRESS_VERSION,
parsePtyStartupIngressIntent
} from './pty-startup-ingress-intent'
export type { PtyStartupIngressIntent } from './pty-startup-ingress-intent'
export type PtyIngressEmission = {
data: string
rawStartSeq: number
rawEndSeq: number
transformed: boolean
}
type PtyIngressSourceChunk = {
data: string
rawStartSeq: number
rawEndSeq: number
}
type PendingOperation =
| { kind: 'data'; chunk: PtyIngressSourceChunk }
| { kind: 'close-query' }
| { kind: 'snapshot' }
| { kind: 'teardown' }
| { kind: 'expire' }
type PendingSpan = PtyIngressSourceChunk
export type PtyStartupIngressOptions = {
intent?: PtyStartupIngressIntent
write: (data: string) => void
onEmission: (emission: PtyIngressEmission) => void
}
const MAX_QUERY_CANDIDATE_CHARS = 64
function spanSlice(span: PendingSpan, start: number, end = span.data.length): PendingSpan {
return {
data: span.data.slice(start, end),
rawStartSeq: span.rawStartSeq + start,
rawEndSeq: span.rawStartSeq + end
}
}
function combineSpans(first: PendingSpan | null, second: PendingSpan): PendingSpan {
if (!first) {
return second
}
return {
data: first.data + second.data,
rawStartSeq: first.rawStartSeq,
rawEndSeq: second.rawEndSeq
}
}
function projectedWindowsConptyReply(reply: string): string {
// Why: the native provider harness observes ConPTY's cooked echo with ESC removed.
return reply.replaceAll('\x1b', '')
}
/**
* Serialized source-side startup classifier. Its raw sequence begins after
* shell-ready preprocessing and every accepted range is emitted exactly once.
*/
export class PtyStartupIngress {
private readonly intent: PtyStartupIngressIntent | undefined
private readonly writeProvider: (data: string) => void
private readonly onEmission: (emission: PtyIngressEmission) => void
private readonly operations: PendingOperation[] = []
private readonly answeredSlots = new Set<TerminalOscColorQuerySlot>()
private readonly expectedEchoes: string[] = []
private processing = false
private closed = false
private queryOpen: boolean
private rawHighWater = 0
private queryPending: PendingSpan | null = null
private echoPending: PendingSpan | null = null
private deadlineTimer: ReturnType<typeof setTimeout> | null = null
constructor(options: PtyStartupIngressOptions) {
this.intent = options.intent
this.writeProvider = options.write
this.onEmission = options.onEmission
this.queryOpen = options.intent !== undefined
if (options.intent) {
this.deadlineTimer = setTimeout(
() => this.enqueue({ kind: 'expire' }),
Math.max(0, options.intent.deadlineMs)
)
this.deadlineTimer.unref?.()
}
}
get acceptedRawSequence(): number {
return this.rawHighWater
}
accept(data: string): void {
if (this.closed || data.length === 0) {
return
}
const rawStartSeq = this.rawHighWater
this.rawHighWater += data.length
this.enqueue({
kind: 'data',
chunk: { data, rawStartSeq, rawEndSeq: this.rawHighWater }
})
}
closeQueryAuthority(): number {
this.enqueue({ kind: 'close-query' })
return this.rawHighWater
}
snapshotBarrier(): number {
this.enqueue({ kind: 'snapshot' })
return this.rawHighWater
}
drainAndClose(): number {
this.enqueue({ kind: 'teardown' })
return this.rawHighWater
}
private enqueue(operation: PendingOperation): void {
if (this.closed) {
return
}
this.operations.push(operation)
if (this.processing) {
return
}
this.processing = true
try {
let next: PendingOperation | undefined
while ((next = this.operations.shift())) {
this.applyOperation(next)
}
} finally {
this.processing = false
}
}
private applyOperation(operation: PendingOperation): void {
switch (operation.kind) {
case 'data':
this.processEchoSpan(operation.chunk)
return
case 'close-query':
this.queryOpen = false
this.releaseQueryPending()
return
case 'expire':
this.queryOpen = false
this.releaseAllPending()
this.expectedEchoes.length = 0
this.clearDeadline()
return
case 'snapshot':
this.releaseSnapshotPending()
return
case 'teardown':
this.queryOpen = false
this.releaseAllPending()
this.expectedEchoes.length = 0
this.clearDeadline()
this.closed = true
}
}
private processEchoSpan(span: PendingSpan): void {
let input = combineSpans(this.echoPending, span)
this.echoPending = null
while (this.expectedEchoes.length > 0) {
const expected = this.expectedEchoes[0]
const compared = Math.min(input.data.length, expected.length)
let matching = 0
while (matching < compared && input.data[matching] === expected[matching]) {
matching += 1
}
if (matching < compared) {
this.expectedEchoes.shift()
this.processQuerySpan(input)
return
}
if (input.data.length < expected.length) {
this.echoPending = input
return
}
this.expectedEchoes.shift()
this.emit(spanSlice(input, 0, expected.length), true, '')
input = spanSlice(input, expected.length)
if (input.data.length === 0) {
return
}
}
this.processQuerySpan(input)
}
private processQuerySpan(span: PendingSpan): void {
const input = combineSpans(this.queryPending, span)
this.queryPending = null
if (!this.queryOpen || !this.intent) {
this.emit(input, false)
return
}
let offset = 0
while (offset < input.data.length) {
const candidateIndex = input.data.indexOf('\x1b', offset)
if (candidateIndex === -1) {
this.emit(spanSlice(input, offset), false)
return
}
if (candidateIndex > offset) {
this.emit(spanSlice(input, offset, candidateIndex), false)
}
const query = parseTerminalOscColorQuery(input.data, candidateIndex)
if (query.kind === 'none') {
this.emit(spanSlice(input, candidateIndex, candidateIndex + 1), false)
offset = candidateIndex + 1
continue
}
if (query.kind === 'partial') {
const candidate = spanSlice(input, candidateIndex)
if (candidate.data.length <= MAX_QUERY_CANDIDATE_CHARS) {
this.queryPending = candidate
} else {
this.emit(candidate, false)
}
return
}
const querySpan = spanSlice(input, candidateIndex, query.endIndex)
if (!this.answerQuery(query.slots)) {
this.emit(querySpan, false)
} else {
this.emit(querySpan, true, '')
}
offset = query.endIndex
}
}
private answerQuery(slots: readonly TerminalOscColorQuerySlot[]): boolean {
if (slots.some((slot) => this.answeredSlots.has(slot)) || !this.intent) {
return false
}
const replies = terminalOscColorQueryReplies(this.intent.colors, slots)
if (!replies) {
return false
}
let wroteAny = false
for (const [index, reply] of replies.entries()) {
const slot = slots[index]
if (slot === undefined) {
return wroteAny
}
this.answeredSlots.add(slot)
const projected =
this.intent.echoProjection === 'windows-conpty-esc-stripped'
? projectedWindowsConptyReply(reply)
: null
if (projected) {
// Why: register before write because node-pty can synchronously re-enter onData.
this.expectedEchoes.push(projected)
}
try {
this.writeProvider(reply)
wroteAny = true
} catch {
this.answeredSlots.delete(slot)
if (projected) {
this.expectedEchoes.pop()
}
return wroteAny
}
}
if (this.answeredSlots.has(10) && this.answeredSlots.has(11)) {
this.queryOpen = false
}
return wroteAny
}
private releaseQueryPending(): void {
if (!this.queryPending) {
return
}
const pending = this.queryPending
this.queryPending = null
this.emit(pending, false)
}
private releaseAllPending(): void {
const pending = this.echoPending ?? this.queryPending
this.echoPending = null
this.queryPending = null
if (pending) {
this.emit(pending, false)
}
}
private releaseSnapshotPending(): void {
if (this.echoPending) {
const pending = this.echoPending
this.echoPending = null
this.expectedEchoes.shift()
this.emit(pending, false)
}
this.releaseQueryPending()
}
private emit(span: PendingSpan, transformed: boolean, data = span.data): void {
this.onEmission({
data,
rawStartSeq: span.rawStartSeq,
rawEndSeq: span.rawEndSeq,
transformed
})
}
private clearDeadline(): void {
if (!this.deadlineTimer) {
return
}
clearTimeout(this.deadlineTimer)
this.deadlineTimer = null
}
}

View File

@ -20,7 +20,8 @@ export enum TerminalStreamOpcode {
Ack = 13,
// Why 14: Ack already occupies 13 on current clients; older runtimes ignore
// this opcode and still receive the compatibility Resize frame behind it.
ClaimViewport = 14
ClaimViewport = 14,
OutputSpan = 15
}
export type TerminalStreamFrame = {
@ -102,6 +103,7 @@ function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode {
value === TerminalStreamOpcode.SnapshotRequest ||
value === TerminalStreamOpcode.Metadata ||
value === TerminalStreamOpcode.Ack ||
value === TerminalStreamOpcode.ClaimViewport
value === TerminalStreamOpcode.ClaimViewport ||
value === TerminalStreamOpcode.OutputSpan
)
}