You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/10/17 07:20:27 UTC
(maka) branch rust-preview updated: feat(terminal): expose ordered raw PTY output
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/rust-preview by this push:
new 6746b3defe feat(terminal): expose ordered raw PTY output
6746b3defe is described below
commit 6746b3defe2b19da7b3635682ed17355ea1f565a
Author: Wang <m4...@gmail.com>
AuthorDate: Sat Oct 17 15:20:13 2026 +0800
feat(terminal): expose ordered raw PTY output
Preserve native bytes before screen decoding and order resize events in the
existing bounded stream. Expose target-scoped reads and subscriptions with
explicit gaps and retention limits while preserving controller custody.
Generated-by: Codex
---
Cargo.lock | 1 +
packages/plugin-sdk/src/contracts.ts | 456 +++++++++++++++++++++
packages/plugin-sdk/src/generated/runtime.ts | 55 +++
platform/host-runtime/src/shell.rs | 2 +-
platform/host-runtime/src/shell/handle.rs | 15 +-
platform/host-runtime/src/shell/output.rs | 401 +++++++++++-------
platform/host-runtime/src/shell/terminal.rs | 60 ++-
platform/host-runtime/src/shell/worker/control.rs | 10 +-
plugins/agent/native/src/owner/tests/terminal.rs | 210 ++++++++++
plugins/workspace/api/README.md | 2 +
plugins/workspace/api/README.zh-CN.md | 2 +
plugins/workspace/api/src/terminal.rs | 66 +++
.../workspace-processes/native/Cargo.toml | 1 +
.../workspace/workspace-processes/native/README.md | 2 +
.../workspace-processes/native/README.zh-CN.md | 2 +
.../workspace-processes/native/src/terminal.rs | 5 +
.../native/src/terminal/output.rs | 106 +++++
products/maka/cli/notices.json | 8 +-
tools/schema/src/lib.rs | 6 +
19 files changed, 1245 insertions(+), 165 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index b54eb56e9d..e8e725401c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8918,6 +8918,7 @@ dependencies = [
name = "maka-processes-native"
version = "0.2.0"
dependencies = [
+ "base64 0.22.1",
"futures-util",
"jsonschema 0.56.0",
"maka-agent-api",
diff --git a/packages/plugin-sdk/src/contracts.ts b/packages/plugin-sdk/src/contracts.ts
index ce17455fd4..0ad2d15d1e 100644
--- a/packages/plugin-sdk/src/contracts.ts
+++ b/packages/plugin-sdk/src/contracts.ts
@@ -37010,6 +37010,234 @@ export const contracts = {
},
resources: {},
},
+ readOutput: {
+ mode: 'query',
+ input: {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ target: {
+ $ref: '#/$defs/TerminalTarget',
+ },
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ default: 0,
+ },
+ },
+ required: ['target'],
+ description: "Cursor belongs to this original target's ordered Data/Resize stream.",
+ title: 'TerminalOutputRequest',
+ $schema: 'https://json-schema.org/draft/2020-12/schema',
+ $defs: {
+ TerminalTarget: {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ sessionId: {
+ type: 'string',
+ },
+ ref: {
+ type: 'string',
+ },
+ },
+ required: ['sessionId', 'ref'],
+ },
+ },
+ },
+ output: {
+ oneOf: [
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ value: {
+ $ref: '#/$defs/TerminalOutputEvent',
+ },
+ kind: {
+ type: 'string',
+ const: 'completed',
+ },
+ },
+ required: ['kind', 'value'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ message: {
+ type: 'string',
+ },
+ kind: {
+ type: 'string',
+ const: 'failed',
+ },
+ },
+ required: ['kind', 'message'],
+ },
+ ],
+ title: 'TerminalOutcome_TerminalOutputEvent',
+ $schema: 'https://json-schema.org/draft/2020-12/schema',
+ $defs: {
+ TerminalOutputEvent: {
+ oneOf: [
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ sequence: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 1,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ dataBase64: {
+ type: 'string',
+ minLength: 1,
+ maxLength: 21848,
+ },
+ kind: {
+ type: 'string',
+ const: 'data',
+ },
+ },
+ required: ['kind', 'sequence', 'size', 'dataBase64'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ sequence: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 1,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ kind: {
+ type: 'string',
+ const: 'resize',
+ },
+ },
+ required: ['kind', 'sequence', 'size'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ },
+ firstAvailable: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 1,
+ maximum: 9007199254740991,
+ },
+ kind: {
+ type: 'string',
+ const: 'gap',
+ },
+ },
+ required: ['kind', 'after', 'firstAvailable'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ kind: {
+ type: 'string',
+ const: 'pending',
+ },
+ },
+ required: ['kind', 'after', 'size'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ kind: {
+ type: 'string',
+ const: 'closed',
+ },
+ },
+ required: ['kind', 'after', 'size'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ reason: {
+ $ref: '#/$defs/TerminalOutputUnavailable',
+ },
+ kind: {
+ type: 'string',
+ const: 'unavailable',
+ },
+ },
+ required: ['kind', 'reason'],
+ },
+ ],
+ description:
+ 'Ephemeral output observation; Closed is not a process cleanup receipt.',
+ },
+ TerminalSize: {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ cols: {
+ type: 'integer',
+ format: 'uint16',
+ minimum: 2,
+ maximum: 240,
+ },
+ rows: {
+ type: 'integer',
+ format: 'uint16',
+ minimum: 1,
+ maximum: 100,
+ },
+ },
+ required: ['cols', 'rows'],
+ description:
+ 'Shared by native PTY allocation, input validation and screen rendering.',
+ },
+ TerminalOutputUnavailable: {
+ type: 'string',
+ enum: ['not_retained'],
+ },
+ },
+ },
+ resources: {},
+ },
release: {
mode: 'mutation',
input: {
@@ -38505,6 +38733,234 @@ export const contracts = {
},
resources: {},
},
+ watchOutput: {
+ mode: 'subscription',
+ input: {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ target: {
+ $ref: '#/$defs/TerminalTarget',
+ },
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ default: 0,
+ },
+ },
+ required: ['target'],
+ description: "Cursor belongs to this original target's ordered Data/Resize stream.",
+ title: 'TerminalOutputRequest',
+ $schema: 'https://json-schema.org/draft/2020-12/schema',
+ $defs: {
+ TerminalTarget: {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ sessionId: {
+ type: 'string',
+ },
+ ref: {
+ type: 'string',
+ },
+ },
+ required: ['sessionId', 'ref'],
+ },
+ },
+ },
+ output: {
+ oneOf: [
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ value: {
+ $ref: '#/$defs/TerminalOutputEvent',
+ },
+ kind: {
+ type: 'string',
+ const: 'completed',
+ },
+ },
+ required: ['kind', 'value'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ message: {
+ type: 'string',
+ },
+ kind: {
+ type: 'string',
+ const: 'failed',
+ },
+ },
+ required: ['kind', 'message'],
+ },
+ ],
+ title: 'TerminalOutcome_TerminalOutputEvent',
+ $schema: 'https://json-schema.org/draft/2020-12/schema',
+ $defs: {
+ TerminalOutputEvent: {
+ oneOf: [
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ sequence: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 1,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ dataBase64: {
+ type: 'string',
+ minLength: 1,
+ maxLength: 21848,
+ },
+ kind: {
+ type: 'string',
+ const: 'data',
+ },
+ },
+ required: ['kind', 'sequence', 'size', 'dataBase64'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ sequence: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 1,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ kind: {
+ type: 'string',
+ const: 'resize',
+ },
+ },
+ required: ['kind', 'sequence', 'size'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ },
+ firstAvailable: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 1,
+ maximum: 9007199254740991,
+ },
+ kind: {
+ type: 'string',
+ const: 'gap',
+ },
+ },
+ required: ['kind', 'after', 'firstAvailable'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ kind: {
+ type: 'string',
+ const: 'pending',
+ },
+ },
+ required: ['kind', 'after', 'size'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ after: {
+ type: 'integer',
+ format: 'uint64',
+ minimum: 0,
+ maximum: 9007199254740991,
+ },
+ size: {
+ $ref: '#/$defs/TerminalSize',
+ },
+ kind: {
+ type: 'string',
+ const: 'closed',
+ },
+ },
+ required: ['kind', 'after', 'size'],
+ },
+ {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ reason: {
+ $ref: '#/$defs/TerminalOutputUnavailable',
+ },
+ kind: {
+ type: 'string',
+ const: 'unavailable',
+ },
+ },
+ required: ['kind', 'reason'],
+ },
+ ],
+ description:
+ 'Ephemeral output observation; Closed is not a process cleanup receipt.',
+ },
+ TerminalSize: {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ cols: {
+ type: 'integer',
+ format: 'uint16',
+ minimum: 2,
+ maximum: 240,
+ },
+ rows: {
+ type: 'integer',
+ format: 'uint16',
+ minimum: 1,
+ maximum: 100,
+ },
+ },
+ required: ['cols', 'rows'],
+ description:
+ 'Shared by native PTY allocation, input validation and screen rendering.',
+ },
+ TerminalOutputUnavailable: {
+ type: 'string',
+ enum: ['not_retained'],
+ },
+ },
+ },
+ resources: {},
+ },
},
},
},
diff --git a/packages/plugin-sdk/src/generated/runtime.ts b/packages/plugin-sdk/src/generated/runtime.ts
index 8e2d776df4..87570497a5 100644
--- a/packages/plugin-sdk/src/generated/runtime.ts
+++ b/packages/plugin-sdk/src/generated/runtime.ts
@@ -288,6 +288,10 @@ export type RuntimeProtocol =
| WorkspaceProcessConfiguration
| TerminalList
| TerminalTarget
+ | TerminalOutputRequest
+ | TerminalOutputEvent
+ | TerminalOutputUnavailable
+ | TerminalOutcome_TerminalOutputEvent
| TerminalStart
| TerminalController
| TerminalControl
@@ -4392,6 +4396,50 @@ export type FormResult =
};
export type QuestionRejection = 'invalid_input' | 'unavailable';
export type WorkspaceProcessSessionKind = 'agent' | 'workhub';
+/**
+ * Ephemeral output observation; Closed is not a process cleanup receipt.
+ */
+export type TerminalOutputEvent =
+ | {
+ sequence: number;
+ size: TerminalSize;
+ dataBase64: string;
+ kind: 'data';
+ }
+ | {
+ sequence: number;
+ size: TerminalSize;
+ kind: 'resize';
+ }
+ | {
+ after: number;
+ firstAvailable: number;
+ kind: 'gap';
+ }
+ | {
+ after: number;
+ size: TerminalSize;
+ kind: 'pending';
+ }
+ | {
+ after: number;
+ size: TerminalSize;
+ kind: 'closed';
+ }
+ | {
+ reason: TerminalOutputUnavailable;
+ kind: 'unavailable';
+ };
+export type TerminalOutputUnavailable = 'not_retained';
+export type TerminalOutcome_TerminalOutputEvent =
+ | {
+ value: TerminalOutputEvent;
+ kind: 'completed';
+ }
+ | {
+ message: string;
+ kind: 'failed';
+ };
export type Ownership =
| {
kind: 'local';
@@ -9969,6 +10017,13 @@ export interface TerminalTarget {
sessionId: string;
ref: string;
}
+/**
+ * Cursor belongs to this original target's ordered Data/Resize stream.
+ */
+export interface TerminalOutputRequest {
+ target: TerminalTarget;
+ after?: number;
+}
export interface TerminalStart {
sessionId: string;
/**
diff --git a/platform/host-runtime/src/shell.rs b/platform/host-runtime/src/shell.rs
index 79f02492d5..411df03243 100644
--- a/platform/host-runtime/src/shell.rs
+++ b/platform/host-runtime/src/shell.rs
@@ -29,7 +29,7 @@ mod worker;
pub use control::Input as ControlInput;
pub use control::{ControlError, ControlErrorKind, WriteReceipt};
pub use handle::{ShellHandle, StopReceipt};
-pub use output::{PtyData, PtyReplay, PtyStream, PtyStreamEvent};
+pub use output::{PtyStream, PtyStreamEvent};
use maka_agent_facts::shell_run::ShellRun;
use maka_agent_facts::terminal::TerminalSize;
diff --git a/platform/host-runtime/src/shell/handle.rs b/platform/host-runtime/src/shell/handle.rs
index 830872ac0a..4212bf0dbb 100644
--- a/platform/host-runtime/src/shell/handle.rs
+++ b/platform/host-runtime/src/shell/handle.rs
@@ -18,7 +18,7 @@
*/
use super::control::{Control, ControlError, Input, WriteReceipt};
-use super::output::{Output, PtyReplay, PtyStream};
+use super::output::{Output, PtyStream};
use super::{ShellError, Update};
use maka_agent_facts::shell_run::ShellRun;
use maka_agent_facts::terminal::TerminalSize;
@@ -72,12 +72,13 @@ impl ShellHandle {
self.control_gate.clone()
}
- pub fn replay(&self) -> Option<PtyReplay> {
- self.pty.as_ref().map(|pty| pty.output.replay())
- }
-
- pub fn attach(&self) -> Option<(PtyReplay, PtyStream)> {
- self.pty.as_ref().map(|pty| pty.output.attach())
+ /// Attach to the original ephemeral byte source; no source means no retained
+ /// raw output, never a fabricated replay of the durable screen projection.
+ pub fn output(&self, after: u64) -> super::Result<Option<PtyStream>> {
+ self.pty
+ .as_ref()
+ .map(|pty| pty.output.stream(after))
+ .transpose()
}
pub fn latest(&self) -> Update {
diff --git a/platform/host-runtime/src/shell/output.rs b/platform/host-runtime/src/shell/output.rs
index 9f9295a743..530ce5aa91 100644
--- a/platform/host-runtime/src/shell/output.rs
+++ b/platform/host-runtime/src/shell/output.rs
@@ -25,38 +25,61 @@ use std::{
};
use tokio::sync::Notify;
-const REPLAY_BYTES: usize = 80 * 1024;
-const FRAME_BYTES: usize = 16 * 1024;
-const FRAME_COUNT: usize = 8;
+const DATA_BYTES: usize = 16 * 1024;
+const EVENT_COUNT: usize = 8;
const MAX_SEQUENCE: u64 = 9_007_199_254_740_991;
-/// An atomic, ephemeral terminal cut, not a durable resource revision.
-#[derive(Clone, Debug)]
-pub struct PtyReplay {
- pub sequence: u64,
- pub buffer: String,
- pub size: TerminalSize,
-}
-
-#[derive(Debug)]
-pub struct PtyData {
- pub sequence: u64,
- pub data: String,
-}
-
-#[derive(Debug)]
+/// Ephemeral native observation order, independent of durable ShellRun revisions.
+#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PtyStreamEvent {
- Data(Arc<PtyData>),
- /// The cursor fell behind the bounded ring. Replace the prior terminal cut.
- Reset(PtyReplay),
- Closed,
+ Data {
+ sequence: u64,
+ size: TerminalSize,
+ bytes: Arc<[u8]>,
+ },
+ Resize {
+ sequence: u64,
+ size: TerminalSize,
+ },
+ /// No cursor advance: resumption after a gap is an explicit caller choice.
+ Gap {
+ after: u64,
+ first_available: u64,
+ },
+ Pending {
+ after: u64,
+ size: TerminalSize,
+ },
+ /// Byte-source closure, not process success or confirmed native cleanup.
+ Closed {
+ after: u64,
+ size: TerminalSize,
+ },
}
struct State {
- replay: PtyReplay,
- frames: VecDeque<Arc<PtyData>>,
+ sequence: u64,
+ size: TerminalSize,
+ // Only consecutive Data/Resize events enter this bounded ring.
+ events: VecDeque<PtyStreamEvent>,
closed: bool,
}
+impl State {
+ fn capacity(&self, count: u64) -> Result<()> {
+ if self.closed || count > MAX_SEQUENCE - self.sequence {
+ return Err(ShellError::Rejected(
+ "PTY output sequence exhausted or closed",
+ ));
+ }
+ Ok(())
+ }
+ fn push(&mut self, event: PtyStreamEvent) {
+ if self.events.len() == EVENT_COUNT {
+ self.events.pop_front();
+ }
+ self.events.push_back(event);
+ }
+}
#[derive(Clone)]
pub(super) struct Output(Arc<Shared>);
@@ -66,22 +89,18 @@ struct Shared {
host_changes: tokio::sync::watch::Sender<()>,
}
-/// Independent observation cursor. Dropping it never cancels its process.
+/// Independent bounded observation cursor. Dropping it never stops the process.
pub struct PtyStream {
output: Output,
- sequence: u64,
+ after: u64,
}
-
impl Output {
pub fn new(size: TerminalSize, host_changes: tokio::sync::watch::Sender<()>) -> Self {
Self(Arc::new(Shared {
state: Mutex::new(State {
- replay: PtyReplay {
- sequence: 0,
- buffer: String::new(),
- size,
- },
- frames: VecDeque::new(),
+ sequence: 0,
+ size,
+ events: VecDeque::new(),
closed: false,
}),
changed: Notify::new(),
@@ -89,93 +108,113 @@ impl Output {
}))
}
- pub fn replay(&self) -> PtyReplay {
- self.0.state.lock().unwrap().replay.clone()
- }
-
- pub fn attach(&self) -> (PtyReplay, PtyStream) {
- let replay = self.replay();
- let sequence = replay.sequence;
- (
- replay,
- PtyStream {
- output: self.clone(),
- sequence,
- },
- )
+ pub fn stream(&self, after: u64) -> Result<PtyStream> {
+ if after > self.0.state.lock().unwrap().sequence {
+ return Err(ShellError::Rejected("PTY output cursor is in the future"));
+ }
+ Ok(PtyStream {
+ output: self.clone(),
+ after,
+ })
}
- /// One parser cut is published atomically. Sequence increments per emitted
- /// frame, not per native read or pre-coalescing chunk.
- pub fn publish(&self, text: &str) -> Result<()> {
- if text.is_empty() {
+ /// Publish original native bytes before any UTF-8 or screen interpretation.
+ pub fn publish(&self, bytes: &[u8]) -> Result<()> {
+ if bytes.is_empty() {
return Ok(());
}
- let chunks = chunks(text);
+ let count = u64::try_from(bytes.len().div_ceil(DATA_BYTES))
+ .map_err(|_| ShellError::Rejected("PTY output sequence exhausted"))?;
let mut state = self.0.state.lock().unwrap();
- if state.closed || state.replay.sequence > MAX_SEQUENCE - chunks.len() as u64 {
- return Err(ShellError::Rejected(
- "PTY output sequence exhausted or closed",
- ));
- }
- state.replay.buffer.push_str(text);
- let excess = state.replay.buffer.len().saturating_sub(REPLAY_BYTES);
- let cut = state.replay.buffer.ceil_char_boundary(excess);
- state.replay.buffer.drain(..cut);
- for data in chunks {
- state.replay.sequence += 1;
- let sequence = state.replay.sequence;
- if state.frames.len() == FRAME_COUNT {
- state.frames.pop_front();
- }
- state.frames.push_back(Arc::new(PtyData {
- sequence,
- data: data.into(),
- }));
+ state.capacity(count)?;
+ for bytes in bytes.chunks(DATA_BYTES) {
+ state.sequence += 1;
+ let event = PtyStreamEvent::Data {
+ sequence: state.sequence,
+ size: state.size,
+ bytes: Arc::from(bytes),
+ };
+ state.push(event);
}
drop(state);
- self.0.changed.notify_waiters();
- self.0.host_changes.send_replace(());
+ self.notify();
Ok(())
}
- pub fn resize(&self, size: TerminalSize) {
- self.0.state.lock().unwrap().replay.size = size;
+ /// The sole worker checks counter capacity before performing native resize.
+ /// This is a preflight, not a second reservation or resize authority.
+ pub fn check_resize(&self, size: TerminalSize) -> Result<()> {
+ let state = self.0.state.lock().unwrap();
+ state.capacity(u64::from(size != state.size))
+ }
+
+ /// Called immediately after native resize succeeds, before parser/SQL work.
+ pub fn resize(&self, size: TerminalSize) -> Result<()> {
+ let mut state = self.0.state.lock().unwrap();
+ state.capacity(u64::from(size != state.size))?;
+ if size == state.size {
+ return Ok(());
+ }
+ state.size = size;
+ state.sequence += 1;
+ let event = PtyStreamEvent::Resize {
+ sequence: state.sequence,
+ size,
+ };
+ state.push(event);
+ drop(state);
+ self.notify();
+ Ok(())
}
pub fn close(&self) {
self.0.state.lock().unwrap().closed = true;
+ self.notify();
+ }
+ fn notify(&self) {
self.0.changed.notify_waiters();
self.0.host_changes.send_replace(());
}
}
-
impl PtyStream {
- pub(crate) fn try_next(&mut self) -> Option<PtyStreamEvent> {
+ /// Immediate observation; only delivered Data/Resize advances this cursor.
+ pub fn read(&mut self) -> PtyStreamEvent {
let state = self.output.0.state.lock().unwrap();
- if self.sequence < state.replay.sequence {
- if let Some(frame) = state
- .frames
- .iter()
- .find(|f| f.sequence == self.sequence + 1)
- {
- self.sequence = frame.sequence;
- return Some(PtyStreamEvent::Data(frame.clone()));
+ if self.after < state.sequence {
+ let first_available = state.sequence - state.events.len() as u64 + 1;
+ if self.after + 1 < first_available {
+ return PtyStreamEvent::Gap {
+ after: self.after,
+ first_available,
+ };
+ }
+ let index = (self.after + 1 - first_available) as usize;
+ let event = state.events[index].clone();
+ self.after += 1;
+ return event;
+ }
+ if state.closed {
+ PtyStreamEvent::Closed {
+ after: self.after,
+ size: state.size,
+ }
+ } else {
+ PtyStreamEvent::Pending {
+ after: self.after,
+ size: state.size,
}
- self.sequence = state.replay.sequence;
- return Some(PtyStreamEvent::Reset(state.replay.clone()));
}
- state.closed.then_some(PtyStreamEvent::Closed)
}
pub async fn next(&mut self) -> PtyStreamEvent {
let output = self.output.clone();
loop {
- // Register before inspecting the cut: no lost wakeup at attach/EOF.
+ // Register before inspection, so publication and EOF cannot be lost.
let changed = output.0.changed.notified();
tokio::pin!(changed);
changed.as_mut().enable();
- if let Some(event) = self.try_next() {
+ let event = self.read();
+ if !matches!(event, PtyStreamEvent::Pending { .. }) {
return event;
}
changed.await;
@@ -183,73 +222,143 @@ impl PtyStream {
}
}
-/// Bound JSON-escaped data, not merely UTF-8: C0 output can expand sixfold.
-/// Frame metadata has its own protocol budget. Eight frames retain <=128 KiB.
-fn chunks(text: &str) -> Vec<&str> {
- let mut chunks = Vec::new();
- let (mut start, mut bytes) = (0, 0);
- for (offset, ch) in text.char_indices() {
- let cost = match ch {
- '"' | '\\' | '\n' | '\r' | '\t' | '\u{08}' | '\u{0c}' => 2,
- '\u{00}'..='\u{1f}' => 6,
- _ => ch.len_utf8(),
- };
- if bytes + cost > FRAME_BYTES {
- chunks.push(&text[start..offset]);
- start = offset;
- bytes = 0;
- }
- bytes += cost;
- }
- if start != text.len() {
- chunks.push(&text[start..]);
- }
- chunks
-}
-
#[cfg(test)]
mod tests {
use super::*;
-
- #[tokio::test]
- async fn attachment_cuts_lag_and_eof_preserve_unicode_and_bounded_json_frames() {
- let output = Output::new(
+ fn output() -> Output {
+ Output::new(
TerminalSize::new(80, 24).unwrap(),
tokio::sync::watch::channel(()).0,
+ )
+ }
+
+ #[test]
+ fn ordered_bytes_and_resize_keep_exact_cuts_and_retained_closed_data() {
+ let output = output();
+ let initial = TerminalSize::new(80, 24).unwrap();
+ let resized = TerminalSize::new(100, 40).unwrap();
+ let mut stream = output.stream(0).unwrap();
+ assert_eq!(
+ stream.read(),
+ PtyStreamEvent::Pending {
+ after: 0,
+ size: initial
+ }
+ );
+ output.publish(b"\xff\0\x1b[31m").unwrap();
+ output.check_resize(resized).unwrap();
+ output.resize(resized).unwrap();
+ output.resize(resized).unwrap(); // same-size acceptance is not a new cut
+ output.publish(&[0xe4, 0xb8, 0xad]).unwrap();
+ output.close();
+ assert_eq!(
+ stream.read(),
+ PtyStreamEvent::Data {
+ sequence: 1,
+ size: initial,
+ bytes: Arc::from(&b"\xff\0\x1b[31m"[..])
+ }
+ );
+ assert_eq!(
+ stream.read(),
+ PtyStreamEvent::Resize {
+ sequence: 2,
+ size: resized
+ }
);
- output.publish("before中").unwrap();
- let (cut, mut stream) = output.attach();
- assert_eq!(cut.buffer, "before中");
- let data = "\0中😀\x1b\\\"".repeat(1800);
- output.publish(&data).unwrap();
- let last = output.replay().sequence;
- let mut actual = String::new();
- for sequence in cut.sequence + 1..=last {
- let PtyStreamEvent::Data(frame) = stream.next().await else {
- panic!("unexpected gap")
+ assert_eq!(
+ stream.read(),
+ PtyStreamEvent::Data {
+ sequence: 3,
+ size: resized,
+ bytes: Arc::from(&[0xe4, 0xb8, 0xad][..])
+ }
+ );
+ assert_eq!(
+ stream.read(),
+ PtyStreamEvent::Closed {
+ after: 3,
+ size: resized
+ }
+ );
+ assert_eq!(
+ stream.read(),
+ PtyStreamEvent::Closed {
+ after: 3,
+ size: resized
+ }
+ );
+ assert!(output.stream(4).is_err());
+ }
+
+ #[test]
+ fn bounded_byte_events_gap_without_advancing_and_explicit_resumption() {
+ let output = output();
+ let mut lagged = output.stream(0).unwrap();
+ let bytes: Vec<u8> = (0..DATA_BYTES * 9 + 7).map(|index| index as u8).collect();
+ output.publish(&bytes).unwrap();
+ let gap = PtyStreamEvent::Gap {
+ after: 0,
+ first_available: 3,
+ };
+ assert_eq!(lagged.read(), gap);
+ output.close();
+ assert_eq!(
+ lagged.read(),
+ gap,
+ "closure must not silently skip an omitted event"
+ );
+ let mut resumed = output.stream(2).unwrap();
+ let mut actual = Vec::new();
+ for expected in 3..=10 {
+ let PtyStreamEvent::Data {
+ sequence, bytes, ..
+ } = resumed.read()
+ else {
+ panic!("expected retained bytes")
};
- assert_eq!(frame.sequence, sequence);
- assert!(serde_json::to_vec(&frame.data).unwrap().len() <= FRAME_BYTES + 2);
- actual.push_str(&frame.data);
+ assert_eq!(sequence, expected);
+ assert!(bytes.len() <= DATA_BYTES);
+ actual.extend_from_slice(&bytes);
}
- assert_eq!(actual, data);
- let (_, mut lagged) = output.attach();
- for _ in 0..20 {
- output.publish(&"中".repeat(4000)).unwrap();
+ assert_eq!(actual, bytes[DATA_BYTES * 2..]);
+ assert!(matches!(
+ resumed.read(),
+ PtyStreamEvent::Closed { after: 10, .. }
+ ));
+ }
+
+ #[tokio::test]
+ async fn resize_only_wakes_a_waiter_and_closure_wakes_without_cursor_advance() {
+ let output = output();
+ let size = TerminalSize::new(100, 40).unwrap();
+ let mut stream = output.stream(0).unwrap();
+ {
+ let next = stream.next();
+ tokio::pin!(next);
+ assert!(futures_util::poll!(next.as_mut()).is_pending());
+ output.resize(size).unwrap();
+ assert_eq!(next.await, PtyStreamEvent::Resize { sequence: 1, size });
}
- let PtyStreamEvent::Reset(cut) = lagged.next().await else {
- panic!("missing reset")
- };
- assert!(cut.buffer.len() <= REPLAY_BYTES);
- assert!(cut.buffer.chars().all(|ch| ch == '中'));
- let pending = lagged.next();
- tokio::pin!(pending);
+ {
+ let next = stream.next();
+ tokio::pin!(next);
+ assert!(futures_util::poll!(next.as_mut()).is_pending());
+ output.close();
+ assert_eq!(next.await, PtyStreamEvent::Closed { after: 1, size });
+ }
+ }
+
+ #[test]
+ fn sequence_capacity_rejects_before_resize_and_never_wraps() {
+ let output = output();
+ output.0.state.lock().unwrap().sequence = MAX_SEQUENCE;
assert!(
- tokio::time::timeout(std::time::Duration::from_millis(1), &mut pending)
- .await
+ output
+ .check_resize(TerminalSize::new(81, 24).unwrap())
.is_err()
);
- output.close();
- assert!(matches!(pending.await, PtyStreamEvent::Closed));
+ assert!(output.publish(b"x").is_err());
+ assert!(output.stream(MAX_SEQUENCE + 1).is_err());
}
}
diff --git a/platform/host-runtime/src/shell/terminal.rs b/platform/host-runtime/src/shell/terminal.rs
index b1fa2d5215..5e96f41409 100644
--- a/platform/host-runtime/src/shell/terminal.rs
+++ b/platform/host-runtime/src/shell/terminal.rs
@@ -50,6 +50,9 @@ impl Terminal {
}
pub fn output(&mut self, bytes: &[u8], eof: bool) -> Result<()> {
+ // Native bytes remain observable even when decoding replaces invalid
+ // text or the parser has failed and final native output is draining.
+ self.output.publish(bytes)?;
let text = self.decoder.decode(bytes, eof);
let Some(screen) = &mut self.screen else {
return Ok(());
@@ -57,9 +60,6 @@ impl Terminal {
match screen.write(&text) {
Ok(replies) => {
self.snapshot = screen.snapshot()?;
- // Observational stream, not a durable outcome. Publish only a
- // completed parser cut, including final output during drain.
- self.output.publish(&text)?;
if !replies.is_empty() {
let pending: usize = self
.writes
@@ -94,7 +94,6 @@ impl Terminal {
.ok_or(ShellError::Rejected("terminal parser unavailable"))?;
screen.resize(size)?;
self.snapshot = screen.snapshot()?;
- self.output.resize(size);
Ok(())
}
@@ -233,6 +232,59 @@ pub(super) async fn drain(
mod tests {
use super::Decoder;
+ #[test]
+ fn raw_output_preserves_native_bytes_before_screen_decoding() {
+ use super::{Output, Terminal};
+ use maka_agent_facts::terminal::TerminalSize;
+
+ let size = TerminalSize::new(80, 24).unwrap();
+ let output = Output::new(size, tokio::sync::watch::channel(()).0);
+ let mut terminal = Terminal::new(size, output.clone());
+ // Exercise the actual Host decoder/parser/ring path. These supplied
+ // native-byte cuts are not a claim of OS PTY transport verification.
+ terminal.output(b"\xff\0\x1b[31m\xe4", false).unwrap();
+ terminal.output(b"\xb8\xad\x1b[0m", false).unwrap();
+ terminal.output(b"\xe2\x82", true).unwrap();
+ let mut stream = output.stream(0).unwrap();
+ let mut raw = Vec::new();
+ while let super::super::output::PtyStreamEvent::Data { bytes, .. } = stream.read() {
+ raw.extend_from_slice(&bytes);
+ }
+ assert_eq!(
+ raw.as_slice(),
+ b"\xff\0\x1b[31m\xe4\xb8\xad\x1b[0m\xe2\x82",
+ "the raw replay must preserve invalid bytes, NUL, ANSI and incomplete EOF independently of the screen decoder"
+ );
+ }
+
+ #[test]
+ fn parser_failure_and_later_drain_preserve_original_raw_bytes() {
+ use super::super::output::PtyStreamEvent;
+ use super::{Output, Terminal};
+ use maka_agent_facts::terminal::TerminalSize;
+
+ let size = TerminalSize::new(80, 24).unwrap();
+ let output = Output::new(size, tokio::sync::watch::channel(()).0);
+ let mut terminal = Terminal::new(size, output.clone());
+ terminal.output(b"\x1b]0;", false).unwrap();
+ let chunk = vec![b'x'; 16 * 1024];
+ for _ in 0..3 {
+ terminal.output(&chunk, false).unwrap();
+ }
+ assert!(terminal.output(&chunk, false).is_err());
+ assert!(terminal.screen.is_none());
+ terminal.output(b"final-drain\xff", true).unwrap();
+ let mut expected = b"\x1b]0;".to_vec();
+ expected.extend_from_slice(&chunk.repeat(4));
+ expected.extend_from_slice(b"final-drain\xff");
+ let mut stream = output.stream(0).unwrap();
+ let mut raw = Vec::new();
+ while let PtyStreamEvent::Data { bytes, .. } = stream.read() {
+ raw.extend_from_slice(&bytes);
+ }
+ assert_eq!(raw, expected);
+ }
+
#[test]
fn every_utf8_cut_matches_lossy_eof_without_losing_terminal_controls() {
let bytes = b"\x1b[31m\xe4\xb8\xad\xf0\x9f\x98\x80\xff\xe2\x82";
diff --git a/platform/host-runtime/src/shell/worker/control.rs b/platform/host-runtime/src/shell/worker/control.rs
index b54c304e31..98a27f2791 100644
--- a/platform/host-runtime/src/shell/worker/control.rs
+++ b/platform/host-runtime/src/shell/worker/control.rs
@@ -70,12 +70,16 @@ impl Worker {
if let Some(size) = control.size {
let result = async {
if size != terminal.snapshot.size {
+ // This sole worker owns publication: preflight the bounded
+ // event counter before accepting an irreversible OS resize.
+ self.output.check_resize(size)?;
child.resize(size).await?;
+ resized = true;
resize_changed = true;
- }
- resized = true;
- if resize_changed {
+ self.output.resize(size)?;
terminal.resize(size)?;
+ } else {
+ resized = true;
}
self.persist(terminal, None).await
}
diff --git a/plugins/agent/native/src/owner/tests/terminal.rs b/plugins/agent/native/src/owner/tests/terminal.rs
index 57f5c56fc6..3737d80691 100644
--- a/plugins/agent/native/src/owner/tests/terminal.rs
+++ b/plugins/agent/native/src/owner/tests/terminal.rs
@@ -184,6 +184,8 @@ async fn terminal_original_receipts_scope_isolation_boundary_revoke_and_read_sto
assert_eq!(started["kind"], "completed", "{started}");
let replayed = mutation(&f, &owner, "start", input, operation).await;
assert_eq!(replayed, started);
+ let pipe_output = query(&f, "readOutput", json!({"target":{"sessionId":session,"ref":started["value"]["result"]["ref"]}})).await;
+ assert_eq!(pipe_output["kind"], "failed", "pipes must not masquerade as PTY raw output: {pipe_output}");
wait_file(&f, "once.txt", "once").await;
files::change(&f, &session, "danger_full_access", false, true).await;
let target = start(&f, &owner, &session).await;
@@ -194,6 +196,8 @@ async fn terminal_original_receipts_scope_isolation_boundary_revoke_and_read_sto
let other = f.create("Other Session").await;
let denied = query(&f, "read", json!({"sessionId":other,"ref":target["ref"]})).await;
assert_eq!(denied["kind"], "failed");
+ let denied_output = query(&f, "readOutput", json!({"target":{"sessionId":other,"ref":target["ref"]}})).await;
+ assert_eq!(denied_output["kind"], "failed", "raw output must keep Session authority: {denied_output}");
let denied = query(&f, "list", json!({"sessionId":"foreign-session"})).await;
assert_eq!(denied["kind"], "failed");
let mut no_caller = f.app.client().subscribe(binding(&f).await,"acquire".into(),target.clone(),scope(&f.issuer,true),Attachments::new().into()).await.unwrap();
@@ -211,6 +215,8 @@ async fn terminal_original_receipts_scope_isolation_boundary_revoke_and_read_sto
assert!(listed["value"]["resources"].as_array().unwrap().iter().all(|r| r["result"]["output"].is_null()));
let foreign = query(&f, "read", json!({"sessionId":session,"ref":"maka://runtime/background-tasks/other-provider-resource"})).await;
assert_eq!(foreign["kind"], "failed");
+ let foreign_output = query(&f, "readOutput", json!({"target":{"sessionId":session,"ref":"maka://runtime/background-tasks/other-provider-resource"}})).await;
+ assert_eq!(foreign_output["kind"], "failed", "raw output must keep provider authority: {foreign_output}");
let (lease, controller) = acquire(&f, &owner, &target).await;
files::change(&f, &session, "read_only", true, true).await;
let denied = mutation(&f,&owner,"control",command(&controller,1,"printf revoked > revoked.txt"),Uuid::new_v4()).await;
@@ -285,3 +291,207 @@ async fn workhub_terminal_selects_coordinator_namespace_and_rejects_original_age
processes::close(&f).await;
}).await.unwrap();
}
+
+fn output_request(target: &Value, after: u64) -> Value {
+ json!({"target":target,"after":after})
+}
+fn output_event(value: Value) -> Value {
+ assert_eq!(value["kind"], "completed", "{value}");
+ assert!(serde_json::to_vec(&value).unwrap().len() <= 48 * 1024);
+ value["value"].clone()
+}
+async fn output_read(f: &Fixture, target: &Value, after: u64) -> Value {
+ output_event(query(f, "readOutput", output_request(target, after)).await)
+}
+fn output_data(event: &Value, after: &mut u64, size: &Value) -> Vec<u8> {
+ use base64::Engine;
+ assert_eq!(event["kind"], "data", "{event}");
+ let sequence = event["sequence"].as_u64().unwrap();
+ assert_eq!(
+ sequence,
+ *after + 1,
+ "raw events must advance in original order: {event}"
+ );
+ assert_eq!(&event["size"], size);
+ *after = sequence;
+ let bytes = base64::engine::general_purpose::STANDARD
+ .decode(event["dataBase64"].as_str().unwrap())
+ .unwrap();
+ assert!(!bytes.is_empty() && bytes.len() <= 16 * 1024);
+ bytes
+}
+async fn output_through(
+ watch: &mut Subscription,
+ after: &mut u64,
+ bytes: &mut Vec<u8>,
+ needle: &[u8],
+ size: &Value,
+) {
+ while !bytes.windows(needle.len()).any(|window| window == needle) {
+ let event = output_event(
+ watch
+ .recv()
+ .await
+ .expect("raw output ended before its ACK fence"),
+ );
+ match event["kind"].as_str().unwrap() {
+ "pending" => assert_eq!(event["after"], *after),
+ "data" => bytes.extend(output_data(&event, after, size)),
+ _ => panic!("unexpected event before raw byte fence: {event}"),
+ }
+ }
+}
+
+#[tokio::test]
+async fn terminal_raw_bytes_resize_resume_gap_and_observer_cancellation_preserve_native_owner() {
+ tokio::time::timeout(Duration::from_secs(90), async {
+ let (_, f, session) = processes::fixture().await;
+ files::change(&f, &session, "danger_full_access", false, true).await;
+ let owner = caller();
+ let target = start(&f, &owner, &session).await;
+ let (lease, controller) = acquire(&f, &owner, &target).await;
+ let denied = f.app.client().start(binding(&f).await, "readOutput".into(), output_request(&target, 0), scope(&f.issuer, false), Attachments::new().into()).await;
+ assert_eq!(denied.err().unwrap(), "denied", "raw queries require the original application grant");
+ let denied = f.app.client().subscribe(binding(&f).await, "watchOutput".into(), output_request(&target, 0), scope(&f.issuer, false), Attachments::new().into()).await;
+ assert_eq!(denied.err().unwrap(), "denied", "raw subscriptions require the original application grant");
+ let original_size = json!({"cols":80,"rows":24});
+ let resized = json!({"cols":100,"rows":40});
+
+ // Only the pre-script interactive prompt is intentionally discarded.
+ // A startup Gap is explicitly resumed before any promised payload exists.
+ let mut after = 0;
+ loop {
+ let event = output_read(&f, &target, after).await;
+ match event["kind"].as_str().unwrap() {
+ "data" | "resize" => after = event["sequence"].as_u64().unwrap(),
+ "gap" => after = event["firstAvailable"].as_u64().unwrap() - 1,
+ "pending" => break,
+ _ => panic!("interactive PTY unexpectedly unavailable: {event}"),
+ }
+ }
+ let mut watch = subscribe(&f, &owner, "watchOutput", output_request(&target, after)).await;
+ // The marker is absent from the entered command, so shell echo/prompt
+ // cannot satisfy this fence. Raw discipline removes echo and ONLCR.
+ std::fs::write(f.workspace.join("raw-fixture.sh"), r#"stty raw -echo
+printf '\036RAW_BEGIN\037\377\000\033[31m\344'
+dd bs=1 count=1 of=/dev/null 2>/dev/null
+printf '\270\255\033[0m'
+stty size
+printf '\036SECOND\037'
+dd bs=1 count=1 of=/dev/null 2>/dev/null
+printf '\342\202'
+"#).unwrap();
+ let launched = mutation(&f, &owner, "control", command(&controller, 1, "exec /bin/sh ./raw-fixture.sh"), Uuid::new_v4()).await;
+ assert_eq!(launched["kind"], "completed", "{launched}");
+ let marker = b"\x1eRAW_BEGIN\x1f";
+ let prefix = b"\xff\0\x1b[31m\xe4";
+ let mut head = marker.to_vec();
+ head.extend_from_slice(prefix);
+ let mut initial = Vec::new();
+ output_through(&mut watch, &mut after, &mut initial, &head, &original_size).await;
+ let at = initial.windows(marker.len()).position(|part| part == marker).unwrap() + marker.len();
+ let mut actual = initial[at..].to_vec();
+ assert_eq!(actual, prefix, "the first split UTF8 byte must already be observable");
+ let before_resize = after;
+ let receipt = mutation(&f, &owner, "control", json!({"controller":controller,"sequence":2,"size":resized}), Uuid::new_v4()).await;
+ assert_eq!(receipt["kind"], "completed", "{receipt}");
+ assert_eq!(receipt["value"]["acceptedBytes"], 0);
+ assert_eq!(receipt["value"]["resizeChanged"], true);
+ let resize = loop {
+ let event = output_event(watch.recv().await.unwrap());
+ if event["kind"] == "pending" { continue; }
+ break event;
+ };
+ assert_eq!(resize["kind"], "resize", "{resize}");
+ assert_eq!(resize["size"], resized);
+ assert_eq!(resize["sequence"], after + 1);
+ assert_eq!(output_read(&f, &target, before_resize).await, resize);
+ after = resize["sequence"].as_u64().unwrap();
+ let future = query(&f, "readOutput", output_request(&target, 9_007_199_254_740_991)).await;
+ assert_eq!(future["kind"], "failed", "known live streams must reject a future sequence: {future}");
+ let ack = mutation(&f, &owner, "control", json!({"controller":controller,"sequence":3,"actions":[{"type":"text","text":"x"}]}), Uuid::new_v4()).await;
+ assert_eq!(ack["kind"], "completed", "{ack}");
+ let middle = b"\xb8\xad\x1b[0m40 100\n\x1eSECOND\x1f";
+ let mut following = Vec::new();
+ output_through(&mut watch, &mut after, &mut following, middle, &resized).await;
+ assert_eq!(following, middle, "stty must witness the actual native resize");
+ actual.extend_from_slice(&following);
+ watch.cancel();
+ watch.completion().await.cleanup.unwrap();
+ assert_eq!(query(&f, "read", target.clone()).await["value"]["result"]["status"], "running");
+
+ // Resize receipts, not timed writes, force nine distinct ordered events
+ // while the real script remains gated on its second ACK.
+ let before_gap = after;
+ for index in 0..9u64 {
+ let receipt = mutation(&f, &owner, "control", json!({"controller":controller,"sequence":4+index,"size":{"cols":101+index,"rows":40}}), Uuid::new_v4()).await;
+ assert_eq!(receipt["kind"], "completed", "observer cancellation must leave its independent controller usable: {receipt}");
+ assert_eq!(receipt["value"]["resizeChanged"], true);
+ }
+ let gap = output_read(&f, &target, before_gap).await;
+ assert_eq!(gap["kind"], "gap", "{gap}");
+ assert_eq!(gap["after"], before_gap);
+ assert!(gap["firstAvailable"].as_u64().unwrap() > before_gap + 1);
+ let mut lost = subscribe(&f, &owner, "watchOutput", output_request(&target, before_gap)).await;
+ assert_eq!(output_event(lost.recv().await.unwrap()), gap);
+ assert!(lost.recv().await.is_none(), "a Gap must finish this watch");
+ let completed = lost.completion().await;
+ completed.cleanup.unwrap();
+ assert_eq!(output_event(completed.result.unwrap()), gap);
+ after = gap["firstAvailable"].as_u64().unwrap() - 1;
+ for col in 102..=109 {
+ let event = output_read(&f, &target, after).await;
+ assert_eq!(event["kind"], "resize", "{event}");
+ assert_eq!(event["sequence"], after + 1);
+ assert_eq!(event["size"], json!({"cols":col,"rows":40}));
+ after += 1;
+ }
+ let final_size = json!({"cols":109,"rows":40});
+ assert_eq!(output_read(&f, &target, after).await, json!({"kind":"pending","after":after,"size":final_size}));
+ let mut resumed = subscribe(&f, &owner, "watchOutput", output_request(&target, after)).await;
+ assert_eq!(output_event(resumed.recv().await.unwrap()), json!({"kind":"pending","after":after,"size":final_size}), "the final observer must be attached before releasing the child");
+ let ack = mutation(&f, &owner, "control", json!({"controller":controller,"sequence":13,"actions":[{"type":"text","text":"y"}]}), Uuid::new_v4()).await;
+ assert_eq!(ack["kind"], "completed", "{ack}");
+ loop {
+ let event = output_event(resumed.recv().await.unwrap());
+ match event["kind"].as_str().unwrap() {
+ "pending" => assert_eq!(event["after"], after),
+ "data" => actual.extend(output_data(&event, &mut after, &final_size)),
+ "closed" => {
+ assert_eq!(event["after"], after);
+ assert_eq!(event["size"], final_size);
+ break;
+ },
+ _ => panic!("unexpected resumed event: {event}"),
+ }
+ }
+ resumed.completion().await.cleanup.unwrap();
+ let mut expected = prefix.to_vec();
+ expected.extend_from_slice(middle);
+ expected.extend_from_slice(b"\xe2\x82");
+ assert_eq!(actual, expected, "raw bytes must retain ff, NUL, ANSI and both split and incomplete UTF8");
+ let mut screen = subscribe(&f, &owner, "watch", target.clone()).await;
+ loop {
+ let snapshot = screen.recv().await.unwrap();
+ let status = snapshot["value"]["result"]["status"].as_str().unwrap();
+ if !matches!(status, "starting" | "running") {
+ assert_eq!(status, "completed", "{snapshot}");
+ break;
+ }
+ }
+ screen.cancel();
+ screen.completion().await.cleanup.unwrap();
+ // Existing stop waits for the real worker's drained fence; it cannot
+ // signal this already completed script or turn raw EOF into exit proof.
+ let stopped = mutation(&f, &owner, "stop", target.clone(), Uuid::new_v4()).await;
+ assert_eq!(stopped["kind"], "completed", "{stopped}");
+ assert_eq!(stopped["value"]["applied"], false);
+ lease.completion().await.cleanup.unwrap();
+ assert_eq!(output_read(&f, &target, after).await, json!({"kind":"unavailable","reason":"not_retained"}));
+ let historical = query(&f, "read", target).await;
+ assert_eq!(historical["kind"], "completed");
+ assert_eq!(historical["value"]["result"]["status"], "completed");
+ assert!(historical["value"]["result"]["output"]["screen"].as_str().unwrap().contains("40 100"));
+ processes::close(&f).await;
+ }).await.unwrap();
+}
diff --git a/plugins/workspace/api/README.md b/plugins/workspace/api/README.md
index d8368fb814..c1a2ef9803 100644
--- a/plugins/workspace/api/README.md
+++ b/plugins/workspace/api/README.md
@@ -23,4 +23,6 @@
Defines the shared contracts for workspace files, processes, terminals, protocol processes and Jev. Native providers and authoring tools use the same declarations. Request paths and workspace identifiers describe a selection; the provider still checks the actual bound directory and call authority.
+The Terminal contract also exposes `readOutput` and `watchOutput` for ordered raw PTY bytes (Base64) and resize events. Each read returns one event from a ring of at most eight events, with up to 16 KiB of bytes per data event. Keep each `after` cursor with its original target. `Gap` ends the watch and reports the first retained event; resume explicitly from `firstAvailable - 1`. A retained suffix cannot restore complete terminal state. Raw output is ephemeral: a new attachment after exi [...]
+
[Source](src/lib.rs) · [Cargo manifest](Cargo.toml)
diff --git a/plugins/workspace/api/README.zh-CN.md b/plugins/workspace/api/README.zh-CN.md
index de77218e28..e23ea15de4 100644
--- a/plugins/workspace/api/README.zh-CN.md
+++ b/plugins/workspace/api/README.zh-CN.md
@@ -23,4 +23,6 @@
定义工作区文件、进程、终端、协议进程和 Jev 的共用契约。原生提供方与开发工具使用相同声明。请求路径和工作区标识用于描述选择；提供方仍需检查实际绑定目录和调用权限。
+Terminal 契约还提供 `readOutput` 和 `watchOutput`，按顺序读取原始 PTY 字节（Base64）及尺寸变化事件。每次读取返回一个事件；缓冲最多保留八个事件，每个数据事件最多包含 16 KiB 字节。`after` 游标必须与原始目标一起保留。`Gap` 会结束订阅并报告最早保留的事件；如需继续，应显式从 `firstAvailable - 1` 恢复。保留的尾段不能还原完整终端状态。原始输出是临时数据：退出或重启后新接入会报告 `Unavailable`，屏幕及历史读取仍可使用。`Closed` 表示输出结束，不代表进程清理已确认。输出观察需要显式方法权限，不需要控制器。
+
[源码](src/lib.rs) · [Cargo 配置](Cargo.toml)
diff --git a/plugins/workspace/api/src/terminal.rs b/plugins/workspace/api/src/terminal.rs
index 32b76d0fe5..b0e1a45ed0 100644
--- a/plugins/workspace/api/src/terminal.rs
+++ b/plugins/workspace/api/src/terminal.rs
@@ -46,6 +46,64 @@ pub struct Target {
#[serde(rename = "ref")]
pub reference: String,
}
+/// Cursor belongs to this original target's ordered Data/Resize stream.
+#[derive(Deserialize, JsonSchema)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+#[schemars(rename = "TerminalOutputRequest")]
+pub struct OutputRequest {
+ pub target: Target,
+ #[serde(default)]
+ #[schemars(range(max = 9007199254740991u64))]
+ pub after: u64,
+}
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, JsonSchema)]
+#[serde(rename_all = "snake_case")]
+#[schemars(rename = "TerminalOutputUnavailable")]
+pub enum OutputUnavailable {
+ NotRetained,
+}
+/// Ephemeral output observation; Closed is not a process cleanup receipt.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, JsonSchema)]
+#[serde(
+ tag = "kind",
+ rename_all = "snake_case",
+ rename_all_fields = "camelCase",
+ deny_unknown_fields
+)]
+#[schemars(rename = "TerminalOutputEvent")]
+pub enum OutputEvent {
+ Data {
+ #[schemars(range(min = 1, max = 9007199254740991u64))]
+ sequence: u64,
+ size: TerminalSize,
+ #[schemars(length(min = 1, max = 21848))]
+ data_base64: String,
+ },
+ Resize {
+ #[schemars(range(min = 1, max = 9007199254740991u64))]
+ sequence: u64,
+ size: TerminalSize,
+ },
+ Gap {
+ #[schemars(range(max = 9007199254740991u64))]
+ after: u64,
+ #[schemars(range(min = 1, max = 9007199254740991u64))]
+ first_available: u64,
+ },
+ Pending {
+ #[schemars(range(max = 9007199254740991u64))]
+ after: u64,
+ size: TerminalSize,
+ },
+ Closed {
+ #[schemars(range(max = 9007199254740991u64))]
+ after: u64,
+ size: TerminalSize,
+ },
+ Unavailable {
+ reason: OutputUnavailable,
+ },
+}
#[derive(Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[schemars(rename = "TerminalStart")]
@@ -152,6 +210,14 @@ pub fn contract() -> Contract {
"watch".into(),
method::<Target, ResourceUpdate>(MethodMode::Subscription),
),
+ (
+ "readOutput".into(),
+ method::<OutputRequest, OutputEvent>(MethodMode::Query),
+ ),
+ (
+ "watchOutput".into(),
+ method::<OutputRequest, OutputEvent>(MethodMode::Subscription),
+ ),
(
"start".into(),
method::<Start, ResourceUpdate>(MethodMode::Mutation),
diff --git a/plugins/workspace/workspace-processes/native/Cargo.toml b/plugins/workspace/workspace-processes/native/Cargo.toml
index 5f7d0af9e7..aa30dc9b01 100644
--- a/plugins/workspace/workspace-processes/native/Cargo.toml
+++ b/plugins/workspace/workspace-processes/native/Cargo.toml
@@ -23,6 +23,7 @@ license.workspace = true
publish.workspace = true
[dependencies]
+base64 = "0.22.1"
maka-agent-facts.workspace = true
maka-capabilities.workspace = true
futures-util.workspace = true
diff --git a/plugins/workspace/workspace-processes/native/README.md b/plugins/workspace/workspace-processes/native/README.md
index e274a078c3..b5843ad2eb 100644
--- a/plugins/workspace/workspace-processes/native/README.md
+++ b/plugins/workspace/workspace-processes/native/README.md
@@ -25,4 +25,6 @@ Provides workspace command, terminal and protocol-process services. `protocol` h
`health` (the default `BackgroundTaskHealth` model tool) reads the original session-owned process record. Logs require `include_logs: true`. An optional HTTP(S) URL is probed by the bound Web service under separate network permission; the process result and endpoint observation are independent. A responding endpoint proves neither listener ownership nor browser readiness. Health calls and their nested probes retain original operation receipts.
+The Terminal [raw output contract](../../api/README.md) uses the original PTY owner and its bounded in-memory ring. Closing an output observer leaves the process and controller unchanged; existing sequenced control and mutation receipts retain their original authority.
+
[Source](src/lib.rs) · [Cargo manifest](Cargo.toml)
diff --git a/plugins/workspace/workspace-processes/native/README.zh-CN.md b/plugins/workspace/workspace-processes/native/README.zh-CN.md
index ef11000a95..2b53c0476c 100644
--- a/plugins/workspace/workspace-processes/native/README.zh-CN.md
+++ b/plugins/workspace/workspace-processes/native/README.zh-CN.md
@@ -25,4 +25,6 @@
`health`（默认模型工具 `BackgroundTaskHealth`）读取当前会话原有的进程记录，只有 `include_logs: true` 才返回日志。可选 HTTP(S) 地址由绑定的 Web 服务在独立网络授权下探测；进程状态与端点观测互相独立，端点响应不能证明监听进程归属或浏览器就绪。健康检查及嵌套探测均保留原始操作回执。
+Terminal [原始输出契约](../../api/README.zh-CN.md)复用原有 PTY owner 及其有界内存缓冲。关闭输出观察不会改变进程或控制器，现有顺序控制和 mutation 回执仍保留原有权威。
+
[源码](src/lib.rs) · [Cargo 配置](Cargo.toml)
diff --git a/plugins/workspace/workspace-processes/native/src/terminal.rs b/plugins/workspace/workspace-processes/native/src/terminal.rs
index 6d1ca41ee2..c3e14acb0d 100644
--- a/plugins/workspace/workspace-processes/native/src/terminal.rs
+++ b/plugins/workspace/workspace-processes/native/src/terminal.rs
@@ -20,6 +20,7 @@
//! Explicit user terminal operations on the original Processes resource owner.
mod controller;
mod launch;
+mod output;
use super::{Processes, SessionConfiguration, cancelled, failed, persistence, worker_error};
pub(super) use controller::Lease;
use futures_util::future::BoxFuture;
@@ -94,6 +95,10 @@ impl Processes {
self.terminal_read(decode(value)?, name == "watch", context)
.await
}
+ "readOutput" | "watchOutput" => {
+ self.terminal_output(decode(value)?, name == "watchOutput", context)
+ .await
+ }
"start" => self
.terminal_start(decode(value)?, context)
.await
diff --git a/plugins/workspace/workspace-processes/native/src/terminal/output.rs b/plugins/workspace/workspace-processes/native/src/terminal/output.rs
new file mode 100644
index 0000000000..562eb11ee9
--- /dev/null
+++ b/plugins/workspace/workspace-processes/native/src/terminal/output.rs
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::*;
+use base64::{Engine, engine::general_purpose::STANDARD};
+use maka_host_runtime::shell::PtyStreamEvent;
+
+impl Processes {
+ pub(super) async fn terminal_output(
+ &self,
+ input: OutputRequest,
+ watch: bool,
+ context: &Context,
+ ) -> Result<Value, ToolError> {
+ if input.after > MAX_SEQUENCE {
+ return Err(failed("Invalid terminal output cursor"));
+ }
+ let record = self.terminal_record(&input.target).await?;
+ if !record.output.is_pty() {
+ return Err(failed("Only a PTY has raw terminal output"));
+ }
+ let mut stream = match self.resources.get(&input.target.session_id, &record.id) {
+ Some(handle) => handle.output(input.after).map_err(failed)?,
+ None => None,
+ };
+ let mut event = match &mut stream {
+ Some(stream) => project(stream.read()),
+ None => OutputEvent::Unavailable {
+ reason: OutputUnavailable::NotRetained,
+ },
+ };
+ let mut last = None;
+ loop {
+ let terminal = matches!(
+ event,
+ OutputEvent::Gap { .. }
+ | OutputEvent::Closed { .. }
+ | OutputEvent::Unavailable { .. }
+ );
+ let value = completed(event)?;
+ maka_plugin_core::json::check_delivery(&value, 48 * 1024).map_err(failed)?;
+ if !watch {
+ return Ok(value);
+ }
+ // The retained output Arc conveys no Session authority of its own.
+ self.terminal_session(&input.target.session_id, false)
+ .await?;
+ tokio::select! { biased;
+ _ = context.scope().cancellation.cancelled() => return last.ok_or_else(|| failed("Process operation cancelled")),
+ _ = self.drain.cancelled() => return last.ok_or_else(|| failed("Process owner is draining")),
+ result = context.emit(value.clone()) => result.map_err(failed)?,
+ }
+ if terminal {
+ return Ok(value);
+ }
+ last = Some(value);
+ // A nonterminal event necessarily came from this original stream.
+ let stream = stream.as_mut().expect("active terminal output stream");
+ event = tokio::select! { biased;
+ _ = context.scope().cancellation.cancelled() => return last.ok_or_else(|| failed("Process operation cancelled")),
+ _ = self.drain.cancelled() => return last.ok_or_else(|| failed("Process owner is draining")),
+ event = stream.next() => project(event),
+ };
+ }
+ }
+}
+
+fn project(event: PtyStreamEvent) -> OutputEvent {
+ match event {
+ PtyStreamEvent::Data {
+ sequence,
+ size,
+ bytes,
+ } => OutputEvent::Data {
+ sequence,
+ size,
+ data_base64: STANDARD.encode(bytes.as_ref()),
+ },
+ PtyStreamEvent::Resize { sequence, size } => OutputEvent::Resize { sequence, size },
+ PtyStreamEvent::Gap {
+ after,
+ first_available,
+ } => OutputEvent::Gap {
+ after,
+ first_available,
+ },
+ PtyStreamEvent::Pending { after, size } => OutputEvent::Pending { after, size },
+ PtyStreamEvent::Closed { after, size } => OutputEvent::Closed { after, size },
+ }
+}
diff --git a/products/maka/cli/notices.json b/products/maka/cli/notices.json
index f755454c7e..b68171cca8 100644
--- a/products/maka/cli/notices.json
+++ b/products/maka/cli/notices.json
@@ -1,12 +1,12 @@
{
"inputs": {
- "Cargo.lock": "8dc81c525e278e03876e1b3715453436714d3565fd23ee7dd285c2a74f1c6f13",
+ "Cargo.lock": "4e1814463750287b0766e24e8d205ca949daa92207a2b72d6dca8360862f56f8",
"Cargo.toml": "9a2a29f608172df146deb8fea723605a6c66c3ca7af4871fffc6cc2509f3cefb",
"package-lock.json": "a5af15aca21fabe0e1bdd48784167df4dc81cee17777d54861dcbd641168a4eb",
"package.json": "38baf3beeb5876849ca16e31868d59bfaee769f260c7555b4a8a1847b92e28a1",
"packages/plugin-sdk/src/configurations.ts": "20e587b47e84a31fc794176f85e7ac71c1b8b7f88a46947f9403011c307fff1e",
- "packages/plugin-sdk/src/contracts.ts": "d1207817a5c183cc99defa614ef73b5534e4eb8a79152e8c1c9abfd406211579",
- "packages/plugin-sdk/src/generated/runtime.ts": "66e0d36196e303b32d8464cdecec86c419da124d9f29ce473e0070855a1b8fe2",
+ "packages/plugin-sdk/src/contracts.ts": "b974da09de4fc8a7977eee8250f9eea1bb402151ff41649eaf0470f1f41ab6d0",
+ "packages/plugin-sdk/src/generated/runtime.ts": "a892e355ce7bd9d1115ecf40cd88d38b330187dfa621c5034f4da3867e1c00a7",
"packages/plugin-sdk/src/index.ts": "27b68be5a29d1857f7fe361c8c678e77ba21413c79ec3870c1ce30d0faf9323f",
"packages/plugin-sdk/src/native.ts": "d29328923ca55874fe1a0fc37985097196d805a09b56043c1a3379fd46bd5a20",
"packages/plugin-sdk/src/ui.ts": "d6c64e9406c41ad6bbda73435b559aad24c31adb3c5bbb4ab6f49fff8a8330ca",
@@ -424,7 +424,7 @@
"plugins/workspace/workspace-jev/engine/Cargo.toml": "35b4c9e8f155e62cff3df30073fff7fd2799a6d513bdb088d67ec4bd50a51dcb",
"plugins/workspace/workspace-jev/native/Cargo.toml": "615f9e482da25590579771cd9be8d0220d83aef61e93d9b8679d480fc5c022c4",
"plugins/workspace/workspace-jev/package.ts": "5f7e270323c49902523f02003b4ecc81434c07ac4d9087c8924cfb062fefd125",
- "plugins/workspace/workspace-processes/native/Cargo.toml": "4a4e8711f2758926c49936bfa6a0da4521f9e7e19ce292db8600ca7fad31b977",
+ "plugins/workspace/workspace-processes/native/Cargo.toml": "524035ea8809a2e648446632d500815ac245f8fa124c93dc24b1952d8cbb7522",
"plugins/workspace/workspace-processes/package.ts": "5c1bdbd1891b9c6ba389de5842c43877aefa816a29dad34c8f43b6e7678ace7d",
"plugins/workspace/workspace-web/api/Cargo.toml": "e5a3a8c5c0fd0219db3b07efb8fae4e700b8bcde8892301a87635adfce615001",
"plugins/workspace/workspace-web/native/Cargo.toml": "e40d175d4ed5300b1c43ecc3af89e9b90c806560b3bda65898a742b39559ca74",
diff --git a/tools/schema/src/lib.rs b/tools/schema/src/lib.rs
index 44ea8eace7..8aba299f77 100644
--- a/tools/schema/src/lib.rs
+++ b/tools/schema/src/lib.rs
@@ -309,6 +309,12 @@ pub fn catalog() -> Value {
root::<maka_workspace_api::processes::Configuration>(&mut g),
root::<maka_workspace_api::terminal::List>(&mut g),
root::<maka_workspace_api::terminal::Target>(&mut g),
+ root::<maka_workspace_api::terminal::OutputRequest>(&mut g),
+ root::<maka_workspace_api::terminal::OutputEvent>(&mut g),
+ root::<maka_workspace_api::terminal::OutputUnavailable>(&mut g),
+ root::<maka_workspace_api::terminal::Outcome<maka_workspace_api::terminal::OutputEvent>>(
+ &mut g,
+ ),
root::<maka_workspace_api::terminal::Start>(&mut g),
root::<maka_workspace_api::terminal::Controller>(&mut g),
root::<maka_workspace_api::terminal::Control>(&mut g),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org