openvibers/openvibe.link Commits

Node: the live screen — screen_stream publishes the machine's screen to its owner's private stream (v0.12.0) (#49)

OP OpenVibers committed 29 days ago · 4d5fc0675b · 12 files, +728 −22

* Node: the live screen — screen_stream publishes the machine's screen to its owner's private stream (v0.12.0)

The live screen (plan T14, "screen streaming to the Console"): a machine's owner watches its screen live on openvibe.bot. Contracts 0.182.0 adds the device kind `screen_stream`.

**Core** (`internal/node/screen.go`):
- `screen_stream` start carries the WHIP URL and publish key of the machine's own private OpenRestream stream, which Bot fills in. The core asks the computer plugin first, since it applies the owner's level and has the capture tool, and never passes it the URL or key. Once the plugin agrees, the core publishes the plugin's frames on a second WHIP session, separate from a robot's camera.
- One stream at a time; a new start replaces the running one.
- A stream ends when:
  - its time is up (30 s to 30 min, 10 min by default);
  - it gets stop;
  - the plugin reports it ended (matched by the stream's id, so a newer stream is never stopped);
  - within 250 ms of the local kill switch or an e-stop.
- A start is refused while a stop is latched.
- The WHIP URL must be https with no credentials in it (http only to loopback, in tests).
- `openvibe-node status` shows `Screen: live … until …` while it runs.

**Computer plugin 0.7.0:**
- Captures a JPEG frame 1 to 5 times a second (2 by default) with the same tools as `screen`.
  - Pillow resizes frames to at most 1600 px wide.
  - Without Pillow, the tool writes JPEG itself: screencapture, grim, import and scrot can; gnome-screenshot is refused.
- Frames go to the core as video with `stream: "screen"`, from a private temporary folder that is removed afterwards.
- Runs at every level, like a screenshot. `allow_screen_stream: false` turns it off.
- Notifies the desktop when it starts.
- Ends on stop, an e-stop or the kill switch.
- Telemetry carries `computer_screen_stream` (`live`/`off`), and `describe` reports `screen.stream`.

**Tests:**
- Go (`screen_test.go`): the real Python plugin with a fake `grim` and a test WHIP receiver.
  - Frames are published with the key as the bearer.
  - Status shows live, and stop ends it (WHIP DELETE).
  - The kill switch ends it within 3 s and refuses a new start.
  - Bad targets are refused: http to a public address, credentials in the URL, no key, an unknown action.
- Python: frames at observe, stop and cleanup, e-stop, replacement and time-up with stream ids, refusals, capability and native JPEG argv.
- `go test ./...` and the plugin tests pass.

Docs: computer.md (The live screen), protocol.md, README.

* Node: the screen stream tests stub the plugin with plain functions (Python 3.9 cannot call a staticmethod object)

Changes

ddiff --git a/README.md b/README.md
iindex 5f55468..64fdd49 100644
---- a/README.md
++++ b/README.md
@@@ -45,6 +45,8 @@ on a community profile runs the same check and warns when it fails. Install a co
  **Your computer for your agents:** `--computer observe|ask|trusted|full` adds the `computer` plugin (terminal, files,
  screen, mouse and keyboard, and a browser of its own for OpenVibe.Actor), running as your own account, at the level you set on the machine; `ask` waits for
  your answer in the tray, a notification's Approve button or `openvibe-node approve <id>` ([docs/computer.md](docs/computer.md)).
++With it on, you can also watch the machine's screen live on openvibe.bot. Only you see it, it runs for up to 30
++minutes, and `openvibe-node stop` ends it at once ([the live screen](docs/computer.md#the-live-screen-screen_stream-node-v0120)).
  
  **A machine that runs your jobs:** add it on openvibe.bot/machines, which prints the same installer with `--worker`: the
  job worker runs each job as an unprivileged user in a sandbox, and the OpenVibe.Codes job runner lets you start a coding
ddiff --git a/cmd/openvibe-node/main.go b/cmd/openvibe-node/main.go
iindex b56908c..d4684d4 100644
---- a/cmd/openvibe-node/main.go
++++ b/cmd/openvibe-node/main.go
@@@ -699,6 +699,14 @@ func cmdStatus(g *globals, asJSON bool, stdout io.Writer) error {
  		fmt.Fprintf(stdout, " (%s)", s.Video.LastError)
  	}
  	fmt.Fprintln(stdout)
++	if s.Screen.State != "" && s.Screen.State != "off" {
++		// Shown only while it runs: the owner sees at a glance that their screen is live.
++		fmt.Fprintf(stdout, "Screen:   %s to your OpenVibe account until %s, %d frames", s.Screen.State, s.Screen.Until, s.Screen.Frames)
++		if s.Screen.LastError != "" && s.Screen.State != "live" {
++			fmt.Fprintf(stdout, " (%s)", s.Screen.LastError)
++		}
++		fmt.Fprintln(stdout, "; `openvibe-node stop` ends it")
++	}
  	fmt.Fprintln(stdout, "Plugins:")
  	for _, p := range s.Plugins {
  		fmt.Fprintf(stdout, "  %-15s %-9s %s %s", p.Name, p.State, p.Driver, p.Version)
ddiff --git a/docs/computer.md b/docs/computer.md
iindex a27a700..ad2b6c5 100644
---- a/docs/computer.md
++++ b/docs/computer.md
@@@ -90,6 +90,24 @@ A program runs in its own process group with the Node account's rights, at most
  from `screencapture` (macOS), `grim` (Wayland) or `import`, `scrot` or `gnome-screenshot` (X11), so the Node must run in
  your graphical session for `screen`; without one the plugin simply does not offer it.
  
++### The live screen (`screen_stream`, Node v0.12.0)
++
++You can watch your machine's screen live on openvibe.bot (**Your machines** → **Show screen**), for 30 seconds to
++30 minutes at a time:
++
++- **Only you see it.** Bot gives your machine's start command the WHIP address and key of its own private
++  OpenRestream stream (no recording, never on Live). No operator or agent supplies them, and agents cannot start it.
++- **How it works.** The plugin captures a JPEG frame with the same tools as `screen`, 1 to 5 times a second (2 by
++  default), at most 1600 px wide when Pillow is installed. The core encodes the frames (ffmpeg, else its built-in
++  encoder) and publishes them on a WebRTC session of their own, separate from a robot's camera.
++- **You are told.** The desktop says when it starts ("Your screen is live on OpenVibe"), and `openvibe-node status`
++  shows `Screen: live … until …` while it runs.
++- **It ends** when its time is up, when you press Stop, when capturing fails five times in a row, and at once on
++  `openvibe-node stop` or an e-stop. While a stop is latched, a new start is refused.
++- **Levels.** It runs at every level, like a screenshot. To never offer it, set `allow_screen_stream: false`.
++- **Without Pillow** the capture tool must write JPEG itself: `screencapture`, `grim`, `import` and `scrot` do;
++  `gnome-screenshot` does not.
++
  **Mouse and keyboard** (`input`) is an acting command like running a program: refused at `observe`, approved one by one
  at `ask` (the approval says exactly what it would do: "click at 640, 360 of a 1280×720 picture of the screen",
  "type 12 characters: 'hello there'", "press ctrl+s"). Coordinates are in the space of the screen picture the agent last
@@@ -121,12 +139,14 @@ text and keys are passed to the tool as arguments, never through a shell, and `k
  | `probe`              | `true`         | `false`: no probes (OpenVibe.Watch's `node` source) |
  | `allow_input`        | `true`         | `false`: never use the mouse or keyboard (no `input` capability) |
  | `allow_browser`      | `true`         | `false`: no browser (no `browser` capability) |
++| `allow_screen_stream` | `true`        | `false`: never show the screen live (`screen.stream` is false) |
  | `browser_path`       | found on PATH  | the browser to drive (Chromium, Chrome, Edge or Brave) |
  | `browser_profile`    | `~/.local/share/openvibe-node/browser` | its profile folder: the agent's own, unless you choose another |
  | `browser_headless`   | `true`         | `false`: show its window |
  | `browser_idle_s`     | 600            | close it after this long without a command |
  
  `describe` reports `terminal`, `files`, `screen`, `input` and `browser` with the level and folders, so Bot shows what the machine allows;
--telemetry carries the sensors `computer_level` (observe while a stop is latched), `computer_waiting` and
--`computer_running`.
++telemetry carries the sensors `computer_level` (observe while a stop is latched), `computer_waiting`,
++`computer_running` and `computer_screen_stream` (`live` or `off`). `screen.stream` says whether the live screen is
++offered.
  Every request is logged by the Node (`journalctl -u openvibe-node`), with what it would do, approved or denied.
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex ca8a8a6..b248baf 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -201,6 +201,7 @@ at most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message
  | `exec`     | `{"argv": ["prog", "arg"…], "cwd"?, "timeout_s"?, "stdin"?}`: run a program on the machine ([computer.md](computer.md)) | computer |
  | `fs`       | `{"op": "list"\|"stat"\|"read"\|"write"\|"mkdir", "path", …}`: a file or folder on the machine                  | computer |
  | `screen`   | `{"max_width"?}`: capture the machine's screen                                                          | computer |
++| `screen_stream` | `{"action": "start", "whip_url", "publish_key", "seconds"?, "fps"?}` or `{"action": "stop"}`: show the screen live on the machine's private stream (Bot fills in the URL and key) | computer |
  | `probe`    | `{"target": "http(s)://…" \| "tcp://host:port", "timeout_s"?}`: a read-only check of the machine's own network | computer |
  
  Sign conventions: `throttle`/`x` positive = forward; `steer` positive = turn right; `y` positive = strafe left;
ddiff --git a/internal/node/computer_test.go b/internal/node/computer_test.go
iindex 9479899..dd0076c 100644
---- a/internal/node/computer_test.go
++++ b/internal/node/computer_test.go
@@@ -26,6 +26,12 @@ import (
  // startComputer runs a Node with only the real Python computer plugin at the given control level, its roots the one
  // folder it returns, against the fake Bot.
  func startComputer(t *testing.T, level string) (*env, string) {
++	t.Helper()
++	return startComputerWith(t, level, nil)
++}
++
++// startComputerWith is startComputer with tweak applied to the plugin's config and the Node's options before it starts.
++func startComputerWith(t *testing.T, level string, tweak func(*config.PluginConfig, *Options)) (*env, string) {
  	t.Helper()
  	if runtime.GOOS == "windows" {
  		t.Skip("end-to-end tests run on unix")
@@@ -66,8 +72,12 @@ func startComputer(t *testing.T, level string) (*env, string) {
  	cfg.Plugins = []config.PluginConfig{{Name: "computer", Config: map[string]any{"level": level, "roots": []string{work}, "timeout_s": 20},
  		Env: map[string]string{"PYTHONPATH": filepath.Join(root, "plugins/sdk") + string(os.PathListSeparator) + filepath.Join(root, "plugins/computer")}}}
  	cfg.Video = config.VideoConfig{Source: "off"}
--	n, err := New(Options{Config: cfg, Paths: paths, Creds: creds, Version: "test", HeartbeatInterval: 100 * time.Millisecond,
--		LinkBackoffMin: 50 * time.Millisecond, LatchPoll: 50 * time.Millisecond, Log: testLog()})
++	opt := Options{Config: cfg, Paths: paths, Creds: creds, Version: "test", HeartbeatInterval: 100 * time.Millisecond,
++		LinkBackoffMin: 50 * time.Millisecond, LatchPoll: 50 * time.Millisecond, Log: testLog()}
++	if tweak != nil {
++		tweak(&opt.Config.Plugins[0], &opt)
++	}
++	n, err := New(opt)
  	if err != nil {
  		t.Fatal(err)
  	}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 0ba716c..07dd54b 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -66,6 +66,8 @@ type Node struct {
  	mgr    *plugins.Manager
  	link   *link.Link
  	pub    *video.Publisher
++	// screen is the owner's live view of the machine's screen (screen.go), apart from a robot's camera.
++	screen screenState
  
  	ctx context.Context
  
@@@ -496,6 +498,10 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, r
  		nack(protocol.FaultBadValue, "value must be a JSON object")
  		return
  	}
++	if c.Kind == protocol.KindScreenStream {
++		n.screenCommand(c, entry, received, nack)
++		return
++	}
  	if protocol.ComputerActs(c.Kind, c.Value) {
  		st := n.latch.State()
  		if st.Local {
@@@ -670,12 +676,20 @@ func (n *Node) output(o plugins.Output) {
  			fields = nil
  		}
  		n.log.Info("plugin event", "plugin", o.Plugin, "event", name)
++		if name == "screen_stream_ended" {
++			sid, _ := o.Data["sid"].(string)
++			n.screenEnded(o.Plugin, sid)
++		}
  		n.trackApproval(o.Plugin, name, o.Data, o.At)
  		n.mu.Lock()
  		n.events = append(n.events, protocol.Event{Name: name, Driver: o.Plugin, TS: o.At.UnixMilli(), Fields: fields})
  		n.mu.Unlock()
  		n.flushTelemetry(true)
  	case plugins.OutVideo:
++		if stream, _ := o.Data["stream"].(string); stream == "screen" {
++			n.screenFrame(o)
++			return
++		}
  		n.mu.Lock()
  		j, from := n.jpeg, n.jpegFrom
  		n.mu.Unlock()
@@@ -875,6 +889,7 @@ type Status struct {
  	Limits    safety.Limits      `json:"limits"`
  	Plugins   []PluginStatus     `json:"plugins"`
  	Video     VideoStatus        `json:"video"`
++	Screen    ScreenStatus       `json:"screen"`
  	ConfigDir string             `json:"config_dir"`
  	StateDir  string             `json:"state_dir"`
  	Telemetry protocol.Telemetry `json:"telemetry"`
@@@ -933,6 +948,7 @@ func (n *Node) Status() Status {
  	if n.worker != nil {
  		s.Jobs = n.worker.Jobs()
  	}
++	s.Screen = n.screenStatus()
  	s.Video.State = video.StateOff
  	if n.pub != nil {
  		s.Video.State, s.Video.Frames, s.Video.LastError = n.pub.State(), n.pub.Frames(), n.pub.LastError()
ddiff --git a/internal/node/screen.go b/internal/node/screen.go
nnew file mode 100644
iindex 0000000..a2467b1
---- /dev/null
++++ b/internal/node/screen.go
@@@ -0,0 +1,264 @@
++package node
++
++import (
++	"context"
++	"encoding/json"
++	"fmt"
++	"net"
++	"net/url"
++	"sync"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
++	"github.com/OpenVibers/OpenVibe.Node/internal/plugins"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++	"github.com/OpenVibers/OpenVibe.Node/internal/safety"
++	"github.com/OpenVibers/OpenVibe.Node/internal/video"
++)
++
++// The screen stream (screen_stream, contracts 0.182.0): the machine's screen shown live to its owner. Bot mints the
++// WHIP URL and publish key of the machine's own private OpenRestream stream and sends them in the start command; the
++// computer plugin captures the frames (the same tools as `screen`, at the owner's level) and emits them as video
++// with stream "screen"; the core encodes and publishes them on a second WHIP session, apart from a robot's camera.
++//
++// A stream ends when its time is up, on stop, when the plugin says it ended, and at once when the local kill switch
++// or an e-stop latches; a start is refused while either is latched. One stream at a time: a new start replaces it.
++const (
++	ScreenDefaultSeconds = 600
++	ScreenMinSeconds     = 30
++	ScreenMaxSeconds     = 1800
++	ScreenDefaultFPS     = 2
++	ScreenMaxFPS         = 5
++	// The built-in encoder's picture size when ffmpeg is missing (ffmpeg keeps the frames' own size).
++	screenFallbackW, screenFallbackH = 960, 544
++)
++
++type screenStream struct {
++	pub    *video.Publisher
++	src    *video.JPEGSource
++	from   string // the plugin whose frames feed it
++	sid    string // the plugin's id for it: its "ended" event names it
++	until  time.Time
++	cancel context.CancelFunc
++	done   chan struct{}
++}
++
++type screenState struct {
++	mu  sync.Mutex
++	cur *screenStream
++}
++
++// ScreenStatus is the screen stream in `openvibe-node status`.
++type ScreenStatus struct {
++	State     string `json:"state"`
++	Until     string `json:"until,omitempty"`
++	Frames    int64  `json:"frames,omitempty"`
++	LastError string `json:"last_error,omitempty"`
++}
++
++type screenValue struct {
++	Action     string `json:"action"`
++	WHIPURL    string `json:"whip_url"`
++	PublishKey string `json:"publish_key"`
++	Seconds    *int   `json:"seconds"`
++	FPS        *int   `json:"fps"`
++}
++
++// screenCommand answers screen_stream: the plugin is asked first (it applies the owner's level and has the capture
++// tool), and only once it said yes does the core publish.
++func (n *Node) screenCommand(c protocol.Command, entry *safety.Entry, received time.Time, nack func(code, msg string)) {
++	var v screenValue
++	if err := json.Unmarshal(c.Value, &v); err != nil {
++		nack(protocol.FaultBadValue, "value must be {action, …}")
++		return
++	}
++	p := n.mgr.Route(protocol.KindScreenStream, c.Target, "")
++	switch v.Action {
++	case "stop":
++		stopped := n.stopScreen("stopped")
++		if p != nil {
++			// Best effort: the plugin stops capturing (its answer does not change the core's).
++			wait := p.Begin(n.ctx, c.ID+":stop", protocol.KindScreenStream, json.RawMessage(`{"action":"stop"}`), 5000)
++			go wait()
++		}
++		data, _ := json.Marshal(map[string]any{"streaming": false, "stopped": stopped})
++		r := protocol.AckData{ID: c.ID, LatencyMS: time.Since(received).Milliseconds(), Data: data}
++		n.dedup.Complete(entry, r)
++		n.send(r)
++		return
++	case "start":
++	default:
++		nack(protocol.FaultBadValue, "action is start or stop")
++		return
++	}
++	st := n.latch.State()
++	if st.Local {
++		nack(protocol.FaultLocalStop, "stopped on the machine with `openvibe-node stop`")
++		return
++	}
++	if st.Remote {
++		nack(protocol.FaultEstopped, st.RemoteReason)
++		return
++	}
++	if err := n.checkScreenURL(v.WHIPURL); err != nil {
++		nack(protocol.FaultBadValue, err.Error())
++		return
++	}
++	if v.PublishKey == "" || len(v.PublishKey) > 256 {
++		nack(protocol.FaultBadValue, "publish_key is required")
++		return
++	}
++	seconds, fps := ScreenDefaultSeconds, ScreenDefaultFPS
++	if v.Seconds != nil {
++		seconds = min(max(*v.Seconds, ScreenMinSeconds), ScreenMaxSeconds)
++	}
++	if v.FPS != nil {
++		fps = min(max(*v.FPS, 1), ScreenMaxFPS)
++	}
++	if p == nil {
++		nack(protocol.FaultUnsupported, "no driver on this device captures the screen (the computer plugin is off)")
++		return
++	}
++	// The plugin never sees the URL or the key.
++	toPlugin, _ := json.Marshal(map[string]any{"action": "start", "seconds": seconds, "fps": fps})
++	wait := p.Begin(n.ctx, c.ID, protocol.KindScreenStream, toPlugin, 15000)
++	key := credentials.NewSecret(v.PublishKey)
++	go func() {
++		r := wait()
++		var msg protocol.Message
++		if r.OK {
++			var got struct {
++				SID string `json:"sid"`
++			}
++			_ = json.Unmarshal(r.Data, &got)
++			until := n.startScreen(v.WHIPURL, key, seconds, fps, p.Name(), got.SID)
++			data, _ := json.Marshal(map[string]any{"streaming": true, "seconds": seconds, "fps": fps, "until_ms": until.UnixMilli()})
++			msg = protocol.AckData{ID: c.ID, LatencyMS: time.Since(received).Milliseconds(), Data: data}
++		} else {
++			msg = protocol.Nack{ID: c.ID, FaultCode: r.FaultCode, Message: r.Message}
++		}
++		n.dedup.Complete(entry, msg)
++		n.send(msg)
++	}()
++}
++
++// checkScreenURL accepts an https WHIP endpoint with no credentials in it (http only to loopback, for tests).
++func (n *Node) checkScreenURL(raw string) error {
++	u, err := url.Parse(raw)
++	if err != nil || u.Host == "" || u.User != nil || len(raw) > 2048 {
++		return fmt.Errorf("whip_url must be an https URL")
++	}
++	switch u.Scheme {
++	case "https":
++		return nil
++	case "http":
++		if ip := net.ParseIP(u.Hostname()); n.opt.VideoLoopback && ip != nil && ip.IsLoopback() {
++			return nil
++		}
++	}
++	return fmt.Errorf("whip_url must be an https URL")
++}
++
++// startScreen publishes the plugin's screen frames until the time is up (replacing a stream that runs) → the end.
++func (n *Node) startScreen(whip string, key credentials.Secret, seconds, fps int, from, sid string) time.Time {
++	n.stopScreen("replaced")
++	ctx, cancel := context.WithCancel(n.ctx)
++	until := time.Now().Add(time.Duration(seconds) * time.Second)
++	src := &video.JPEGSource{FFmpeg: n.opt.Config.Video.FFmpeg, FPS: fps, W: screenFallbackW, H: screenFallbackH, Log: n.log}
++	pub := video.NewPublisher(video.Options{WHIPURL: whip, PublishKey: key, ICEServers: n.screenICE(), Log: n.log, IncludeLoopback: n.opt.VideoLoopback})
++	s := &screenStream{pub: pub, src: src, from: from, sid: sid, until: until, cancel: cancel, done: make(chan struct{})}
++	n.screen.mu.Lock()
++	n.screen.cur = s
++	n.screen.mu.Unlock()
++	n.log.Info("screen stream started", "seconds", seconds, "fps", fps)
++	go func() {
++		defer close(s.done)
++		pub.Run(ctx, src)
++	}()
++	go func() {
++		t := time.NewTicker(250 * time.Millisecond)
++		defer t.Stop()
++		for {
++			select {
++			case <-ctx.Done():
++				return
++			case now := <-t.C:
++				st := n.latch.State()
++				switch {
++				case st.Local || st.Remote:
++					n.endScreen(s, "stopped on the machine")
++					return
++				case now.After(until.Add(3 * time.Second)):
++					n.endScreen(s, "time is up")
++					return
++				}
++			}
++		}
++	}()
++	return until
++}
++
++func (n *Node) screenICE() []protocol.ICEServer {
++	if n.opt.Creds == nil {
++		return nil
++	}
++	return n.opt.Creds.ICEServers
++}
++
++// endScreen ends s if it is still the running stream.
++func (n *Node) endScreen(s *screenStream, why string) {
++	n.screen.mu.Lock()
++	if n.screen.cur != s {
++		n.screen.mu.Unlock()
++		return
++	}
++	n.screen.cur = nil
++	n.screen.mu.Unlock()
++	s.cancel()
++	n.log.Info("screen stream ended", "why", why)
++}
++
++// stopScreen ends the running stream, if any → whether one ran.
++func (n *Node) stopScreen(why string) bool {
++	n.screen.mu.Lock()
++	s := n.screen.cur
++	n.screen.mu.Unlock()
++	if s == nil {
++		return false
++	}
++	n.endScreen(s, why)
++	return true
++}
++
++// screenEnded ends the stream the plugin says it ended (by its id), never a newer one.
++func (n *Node) screenEnded(plugin, sid string) {
++	n.screen.mu.Lock()
++	s := n.screen.cur
++	n.screen.mu.Unlock()
++	if s != nil && s.from == plugin && s.sid != "" && s.sid == sid {
++		n.endScreen(s, "the plugin ended it")
++	}
++}
++
++// screenFrame passes a plugin's screen frame to the running stream (frames from any other plugin are dropped).
++func (n *Node) screenFrame(o plugins.Output) {
++	n.screen.mu.Lock()
++	s := n.screen.cur
++	n.screen.mu.Unlock()
++	if s == nil || s.from != o.Plugin {
++		return
++	}
++	if p, ok := o.Data["path"].(string); ok {
++		s.src.Push(p)
++	}
++}
++
++func (n *Node) screenStatus() ScreenStatus {
++	n.screen.mu.Lock()
++	s := n.screen.cur
++	n.screen.mu.Unlock()
++	if s == nil {
++		return ScreenStatus{State: video.StateOff}
++	}
++	return ScreenStatus{State: s.pub.State(), Until: s.until.UTC().Format(time.RFC3339), Frames: s.pub.Frames(), LastError: s.pub.LastError()}
++}
ddiff --git a/internal/node/screen_test.go b/internal/node/screen_test.go
nnew file mode 100644
iindex 0000000..abe17fa
---- /dev/null
++++ b/internal/node/screen_test.go
@@@ -0,0 +1,140 @@
++//go:build !windows
++
++package node
++
++import (
++	"image"
++	"image/color"
++	"image/jpeg"
++	"net/http/httptest"
++	"os"
++	"path/filepath"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/localctl"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++	"github.com/OpenVibers/OpenVibe.Node/internal/video"
++)
++
++// startScreenNode runs the real computer plugin with a fake `grim` (a Wayland session) that copies a real JPEG to the
++// file it is given, so the screen stream has frames without a display; video may use loopback.
++func startScreenNode(t *testing.T, level string) *env {
++	t.Helper()
++	bin := t.TempDir()
++	pic := filepath.Join(bin, "screen.jpg")
++	img := image.NewRGBA(image.Rect(0, 0, 320, 240))
++	for y := 0; y < 240; y++ {
++		for x := 0; x < 320; x++ {
++			img.Set(x, y, color.RGBA{uint8(x), uint8(y), 160, 255})
++		}
++	}
++	f, err := os.Create(pic)
++	if err != nil {
++		t.Fatal(err)
++	}
++	if err := jpeg.Encode(f, img, nil); err != nil {
++		t.Fatal(err)
++	}
++	f.Close()
++	// grim [-t jpeg -q N] <out>: the output is the last argument.
++	script := "#!/bin/sh\nfor last; do :; done\ncp \"" + pic + "\" \"$last\"\n"
++	if err := os.WriteFile(filepath.Join(bin, "grim"), []byte(script), 0o755); err != nil {
++		t.Fatal(err)
++	}
++	e, _ := startComputerWith(t, level, func(pc *config.PluginConfig, o *Options) {
++		pc.Env["PATH"] = bin + string(os.PathListSeparator) + os.Getenv("PATH")
++		pc.Env["WAYLAND_DISPLAY"] = "wayland-test"
++		pc.Config["notify"] = false
++		o.VideoLoopback = true
++	})
++	return e
++}
++
++func screenWHIP(t *testing.T) (*video.WHIPReceiver, string) {
++	t.Helper()
++	w := &video.WHIPReceiver{PublishKey: "ork_screen_test_key"}
++	srv := httptest.NewServer(w)
++	t.Cleanup(func() { w.Close(); srv.Close() })
++	return w, srv.URL + "/whip/ork_screen_test_key"
++}
++
++func TestScreenStreamPublishesTheScreenAndStops(t *testing.T) {
++	e := startScreenNode(t, "observe")
++	whip, url := screenWHIP(t)
++	r := e.computerCmd("s1", protocol.KindScreenStream, map[string]any{"action": "start", "whip_url": url, "publish_key": "ork_screen_test_key", "seconds": 60, "fps": 5}, 20*time.Second)
++	d := dataOf(t, r)
++	if d["streaming"] != true || d["seconds"] != float64(60) || d["fps"] != float64(5) {
++		t.Fatalf("%+v", d)
++	}
++	deadline := time.Now().Add(20 * time.Second)
++	for whip.Packets.Load() < 10 {
++		if time.Now().After(deadline) {
++			t.Fatalf("no screen video: %+v", e.node.Status().Screen)
++		}
++		time.Sleep(50 * time.Millisecond)
++	}
++	if st := e.node.Status().Screen; st.State != video.StateLive || st.Until == "" {
++		t.Fatalf("status %+v", st)
++	}
++	if whip.BadAuth.Load() != 0 {
++		t.Fatal("the publish key was not sent as the bearer")
++	}
++	stop := dataOf(t, e.computerCmd("s2", protocol.KindScreenStream, map[string]any{"action": "stop"}, 10*time.Second))
++	if stop["stopped"] != true {
++		t.Fatalf("%+v", stop)
++	}
++	deadline = time.Now().Add(5 * time.Second)
++	for e.node.Status().Screen.State != video.StateOff || whip.Deletes.Load() == 0 {
++		if time.Now().After(deadline) {
++			t.Fatalf("still streaming after stop: %+v (deletes %d)", e.node.Status().Screen, whip.Deletes.Load())
++		}
++		time.Sleep(50 * time.Millisecond)
++	}
++}
++
++func TestScreenStreamEndsOnTheKillSwitchAndIsRefusedWhileLatched(t *testing.T) {
++	e := startScreenNode(t, "trusted")
++	whip, url := screenWHIP(t)
++	start := map[string]any{"action": "start", "whip_url": url, "publish_key": "ork_screen_test_key", "fps": 5}
++	dataOf(t, e.computerCmd("k1", protocol.KindScreenStream, start, 20*time.Second))
++	deadline := time.Now().Add(20 * time.Second)
++	for whip.Packets.Load() < 5 {
++		if time.Now().After(deadline) {
++			t.Fatalf("no screen video: %+v", e.node.Status().Screen)
++		}
++		time.Sleep(50 * time.Millisecond)
++	}
++	if _, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStop}, 2*time.Second); err != nil {
++		t.Fatal(err)
++	}
++	deadline = time.Now().Add(3 * time.Second)
++	for e.node.Status().Screen.State != video.StateOff {
++		if time.Now().After(deadline) {
++			t.Fatalf("the kill switch did not end the stream: %+v", e.node.Status().Screen)
++		}
++		time.Sleep(25 * time.Millisecond)
++	}
++	if c := nackCode(e.computerCmd("k2", protocol.KindScreenStream, start, 10*time.Second)); c != protocol.FaultLocalStop {
++		t.Fatalf("start while latched: %q", c)
++	}
++}
++
++func TestScreenStreamRefusesBadTargets(t *testing.T) {
++	e := startScreenNode(t, "observe")
++	for i, v := range []map[string]any{
++		{"action": "start", "whip_url": "http://203.0.113.5/whip/k", "publish_key": "k"},
++		{"action": "start", "whip_url": "https://user:[email protected]/whip/k", "publish_key": "k"},
++		{"action": "start", "whip_url": "https://openre.stream/whip/k"},
++		{"action": "watch"},
++	} {
++		r := e.computerCmd("b"+string(rune('0'+i)), protocol.KindScreenStream, v, 10*time.Second)
++		if c := nackCode(r); c != protocol.FaultBadValue {
++			t.Fatalf("%v: %q", v, c)
++		}
++	}
++	if st := e.node.Status().Screen.State; st != video.StateOff {
++		t.Fatalf("%q", st)
++	}
++}
ddiff --git a/internal/plugins/manager.go b/internal/plugins/manager.go
iindex 27ad58e..2291466 100644
---- a/internal/plugins/manager.go
++++ b/internal/plugins/manager.go
@@@ -10,19 +10,20 @@ import (
  
  // capabilityFor maps a command kind to the describe capability key that declares it.
  var capabilityFor = map[string]string{
--	protocol.KindDrive:    "drive",
--	protocol.KindPTZ:      "ptz",
--	protocol.KindActuator: "actuator",
--	protocol.KindSay:      "say",
--	protocol.KindDisplay:  "display",
--	protocol.KindButton:   "button",
--	protocol.KindPoint:    "point",
--	protocol.KindExec:     "terminal",
--	protocol.KindFS:       "files",
--	protocol.KindScreen:   "screen",
--	protocol.KindProbe:    "probe",
--	protocol.KindInput:    "input",
--	protocol.KindBrowser:  "browser",
++	protocol.KindDrive:        "drive",
++	protocol.KindPTZ:          "ptz",
++	protocol.KindActuator:     "actuator",
++	protocol.KindSay:          "say",
++	protocol.KindDisplay:      "display",
++	protocol.KindButton:       "button",
++	protocol.KindPoint:        "point",
++	protocol.KindExec:         "terminal",
++	protocol.KindFS:           "files",
++	protocol.KindScreen:       "screen",
++	protocol.KindScreenStream: "screen",
++	protocol.KindProbe:        "probe",
++	protocol.KindInput:        "input",
++	protocol.KindBrowser:      "browser",
  }
  
  // Manager supervises every configured plugin.
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex f4db54e..9cb7cb7 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -170,13 +170,18 @@ const (
  	// KindBrowser drives a Chromium the agent uses, in a profile of its own (contracts 0.174.0): open, click, fill,
  	// press and back act (the owner's level decides, a latched stop refuses them); read, screenshot and close do not.
  	KindBrowser = "browser"
++	// KindScreenStream shows the machine's screen live to its owner (contracts 0.182.0): {"action": "start",
++	// "whip_url", "publish_key", "seconds"?, "fps"?} or {"action": "stop"}. Bot fills in the WHIP URL and key (the
++	// machine's private OpenRestream stream); no operator sends them. The computer plugin captures the frames, the
++	// core publishes them; a latched stop ends it and refuses a start.
++	KindScreenStream = "screen_stream"
  )
  
  // Kinds lists every command kind the Node understands.
--var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt, KindButton, KindPoint, KindExec, KindFS, KindScreen, KindProbe, KindInput, KindBrowser}
++var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt, KindButton, KindPoint, KindExec, KindFS, KindScreen, KindProbe, KindInput, KindBrowser, KindScreenStream}
  
  // ComputerKinds move nothing: they carry long deadlines (DefaultComputerMS, at most MaxComputerMS) and no clamping.
--var ComputerKinds = map[string]bool{KindExec: true, KindFS: true, KindScreen: true, KindProbe: true, KindInput: true, KindBrowser: true}
++var ComputerKinds = map[string]bool{KindExec: true, KindFS: true, KindScreen: true, KindProbe: true, KindInput: true, KindBrowser: true, KindScreenStream: true}
  
  const (
  	// DefaultComputerMS is a computer command's reply window when the server sends no deadline; MaxComputerMS caps
ddiff --git a/plugins/computer/openvibe_computer/__init__.py b/plugins/computer/openvibe_computer/__init__.py
iindex 360f284..19c82c3 100644
---- a/plugins/computer/openvibe_computer/__init__.py
++++ b/plugins/computer/openvibe_computer/__init__.py
@@@ -21,6 +21,12 @@ Command kinds and their answers (the ack's ``data``):
  * ``fs`` ``{"op": "list"|"stat"|"read"|"write"|"mkdir", "path", …}``: ``read`` takes ``offset``/``length`` and answers
    ``content_b64``; ``write`` takes ``content_b64`` and ``mode`` (``create``, ``overwrite`` or ``append``).
  * ``screen`` ``{"max_width"?}`` → ``{format, width, height, content_b64}`` (JPEG through Pillow when it is installed).
++* ``screen_stream`` ``{"action": "start", "seconds"?, "fps"?}`` → ``{streaming, seconds, fps, tool}``, or ``{"action":
++  "stop"}``: the screen shown live to the machine's owner. The plugin captures a JPEG frame ``fps`` times a second (at
++  most 5) with the same tools as ``screen`` and hands each to the core as video (``stream: "screen"``), which publishes
++  it to the owner's private stream; the core alone holds the WHIP URL and key. It runs at every level, like a
++  screenshot, says so on the desktop when it starts (config ``notify``), and ends when its time is up (30 s to 30 min),
++  on stop, and at once on the local kill switch or an e-stop. Config ``allow_screen_stream: false`` turns it off.
  * ``input`` ``{"action", "x"?, "y"?, "width"?, "height"?, "dy"?, "text"?, "keys"?}`` → ``{done, action, x?, y?}``: move,
    click, double_click or right_click at a point, scroll by ``dy`` clicks, type ``text`` into whatever has focus, or press
    one key or chord (``keys``, for example ``ctrl+s``). ``x``/``y`` are in the space of a screen picture ``width`` ×
@@@ -79,6 +85,11 @@ INPUT_GUARD_S = 2.0
  MAX_OUTPUT_CAP = 192 * 1024
  MAX_READ_CAP = 256 * 1024
  SCREEN_MAX_BYTES = 300 * 1024
++# The live screen: frame width (Pillow resizes larger screens) and JPEG quality; the stream ends after this many
++# capture failures in a row.
++STREAM_WIDTH = 1600
++STREAM_QUALITY = 70
++STREAM_MAX_FAILURES = 5
  INPUT_ACTIONS = frozenset({"move", "click", "double_click", "right_click", "scroll", "type", "key"})
  BROWSER_READS = frozenset({"read", "screenshot", "close"})
  BROWSER_ACTS = frozenset({"open", "click", "fill", "press", "back"})
@@@ -121,7 +132,7 @@ def _int(v, name, lo, hi, default):
  
  class Computer(Plugin):
      driver = "computer"
--    version = "0.6.0"
++    version = "0.7.0"
      motion_kinds = frozenset()
      telemetry_interval_s = 2.0
  
@@@ -158,6 +169,9 @@ class Computer(Plugin):
          self.browser_headless = True
          self.browser_idle_s = 600
          self.browser = None
++        self.allow_screen_stream = True
++        self.stream = None  # the live screen: {"sid", "until", "fps", "tool", "dir", "stop": Event}
++        self.stream_mu = threading.Lock()
          self.browser_mu = threading.Lock()
          self.browser_op_mu = threading.Lock()
          self.browser_factory = browser_factory or (lambda binary, profile, headless: Browser(binary, profile, headless))
@@@ -184,6 +198,7 @@ class Computer(Plugin):
          self.allow_input = config.get("allow_input", True) is not False
          self.notify = config.get("notify", True) is not False
          self.allow_browser = config.get("allow_browser", True) is not False
++        self.allow_screen_stream = config.get("allow_screen_stream", True) is not False
          self.browser_path = config.get("browser_path") or None
          self.browser_profile = os.path.expanduser(config.get("browser_profile") or "~/.local/share/openvibe-node/browser")
          self.browser_headless = config.get("browser_headless", True) is not False
@@@ -199,7 +214,8 @@ class Computer(Plugin):
              caps["terminal"] = {"level": self.level, "timeout_s": self.timeout_s, "max_output_bytes": self.max_output}
          tool = self._screen_tool()
          if tool:
--            caps["screen"] = {"level": self.level, "tool": tool[0]}
++            caps["screen"] = {"level": self.level, "tool": tool[0],
++                              "stream": bool(self.allow_screen_stream and self._stream_capable(tool))}
          if self.allow_probe:
              caps["probe"] = {"lan_only": True, "schemes": ["http", "https", "tcp"]}
          itool = self._input_tool() if self.allow_input else None
@@@ -231,6 +247,7 @@ class Computer(Plugin):
          for p in procs:
              self._kill(p)
          self._close_browser()
++        self._end_stream("stopped on the machine")
          for cid in waiting:
              self._close_prompt(cid)
              self.emit_event("approval_done", id=cid, approved=False, reason="stopped")
@@@ -262,7 +279,9 @@ class Computer(Plugin):
              self.emit_event("approval_done", id=cid, approved=False, reason="expired")
              self.fail(cid, "not_allowed", "nobody answered on the machine in time")
          # Sensors are what the core forwards to Bot: the machines page shows the level and what waits on the machine.
--        return {"sensors": {"computer_level": self.effective_level(), "computer_waiting": waiting, "computer_running": running}}
++        streaming = self.stream is not None
++        return {"sensors": {"computer_level": self.effective_level(), "computer_waiting": waiting, "computer_running": running,
++                            "computer_screen_stream": "live" if streaming else "off"}}
  
      # ---- paths -----------------------------------------------------------------------------------------------------
      def _inside(self, path):
@@@ -296,6 +315,8 @@ class Computer(Plugin):
                  raise Fault("unsupported", "no screen capture tool on this machine")
              width = _int(v.get("max_width"), "max_width", 320, 3840, 1280)
              return self._start(cmd, lambda: self._screen(tool, width))
++        if cmd.kind == "screen_stream":
++            return self._screen_stream(v)
          if cmd.kind == "fs":
              op = v.get("op")
              if op in READ_OPS:
@@@ -1041,6 +1062,149 @@ class Computer(Plugin):
              if os.path.exists(out):
                  os.unlink(out)
  
++    # ---- the live screen --------------------------------------------------------------------------------------------
++    @staticmethod
++    def _have_pillow():
++        try:
++            import PIL.Image  # noqa: F401
++        except ImportError:
++            return False
++        return True
++
++    @staticmethod
++    def _native_jpeg(tool, path):
++        """The capture tool writing JPEG itself (for machines without Pillow), or None when it only writes PNG."""
++        name = tool[0]
++        if name == "screencapture":
++            return ["screencapture", "-x", "-t", "jpg", path]
++        if name == "grim":
++            return ["grim", "-t", "jpeg", "-q", str(STREAM_QUALITY), path]
++        if name == "import":
++            return ["import", "-silent", "-window", "root", "-quality", str(STREAM_QUALITY), path]
++        if name == "scrot":
++            return ["scrot", "-o", "-q", str(STREAM_QUALITY), path]
++        return None
++
++    def _stream_capable(self, tool):
++        return self._have_pillow() or self._native_jpeg(tool, "x.jpg") is not None
++
++    def _screen_stream(self, v):
++        action = v.get("action")
++        if action == "stop":
++            return {"streaming": False, "stopped": self._end_stream("stopped by its owner")}
++        if action != "start":
++            raise Fault("bad_value", "action is start or stop")
++        if not self.allow_screen_stream:
++            raise Fault("unsupported", "this machine does not show its screen live (allow_screen_stream is off)")
++        if self.runtime is not None and self.runtime.estopped:
++            raise Fault("local_stop", "stopped on the machine")
++        tool = self._screen_tool()
++        if not tool:
++            raise Fault("unsupported", "no screen capture tool on this machine")
++        if not self._stream_capable(tool):
++            raise Fault("unsupported", "%s only writes PNG: install Pillow (pip install pillow) to show the screen live" % tool[0])
++        seconds = _int(v.get("seconds"), "seconds", 30, 1800, 600)
++        fps = _int(v.get("fps"), "fps", 1, 5, 2)
++        self._end_stream("replaced")
++        frames = tempfile.mkdtemp(prefix="openvibe-screen-stream-")  # 0700, this account only
++        stream = {"sid": os.urandom(8).hex(), "until": self.clock() + seconds, "fps": fps, "tool": tool, "dir": frames,
++                  "stop": threading.Event()}
++        with self.stream_mu:
++            self.stream = stream
++        threading.Thread(target=self._stream_loop, args=(stream,), name="computer-screen-stream", daemon=True).start()
++        self._notify_stream(seconds)
++        return {"streaming": True, "sid": stream["sid"], "seconds": seconds, "fps": fps, "tool": tool[0]}
++
++    def _end_stream(self, reason):
++        """End the live screen, if one runs → whether one did."""
++        with self.stream_mu:
++            stream, self.stream = self.stream, None
++        if stream is None:
++            return False
++        stream["stop"].set()
++        stream["reason"] = reason
++        return True
++
++    def _stream_loop(self, stream):
++        period = 1.0 / stream["fps"]
++        n, failures, reason = 0, 0, "time is up"
++        try:
++            while not stream["stop"].is_set():
++                started = self.clock()
++                if started >= stream["until"]:
++                    break
++                path = os.path.join(stream["dir"], "frame-%d.jpg" % (n % 3))  # the core reads each one at once
++                try:
++                    self._stream_frame(stream["tool"], path)
++                    n += 1
++                    failures = 0
++                    self.emit_video(path, stream="screen")
++                except Fault as f:
++                    failures += 1
++                    if failures >= STREAM_MAX_FAILURES:
++                        reason = "the screen capture failed: %s" % f.message
++                        break
++                stream["stop"].wait(max(0.0, period - (self.clock() - started)))
++        finally:
++            shutil.rmtree(stream["dir"], ignore_errors=True)
++            with self.stream_mu:
++                if self.stream is stream:
++                    self.stream = None
++            # The core ends the stream with the same id (a newer one that replaced it carries its own).
++            self.emit_event("screen_stream_ended", sid=stream["sid"], reason=stream.get("reason") or reason, frames=n)
++
++    def _stream_frame(self, tool, path):
++        """One JPEG frame of the screen at path, at most STREAM_WIDTH wide when Pillow can resize it."""
++        if self._have_pillow():
++            fd, png = tempfile.mkstemp(prefix="openvibe-screen-", suffix=".png")
++            os.close(fd)
++            try:
++                self._capture(tool[1](png))
++                from PIL import Image
++                with Image.open(png) as im:
++                    im = im.convert("RGB")
++                    if im.width > STREAM_WIDTH:
++                        im = im.resize((STREAM_WIDTH, max(1, round(im.height * STREAM_WIDTH / im.width))))
++                    im.save(path + ".part", "JPEG", quality=STREAM_QUALITY)
++                os.replace(path + ".part", path)
++            finally:
++                if os.path.exists(png):
++                    os.unlink(png)
++            return
++        self._capture(self._native_jpeg(tool, path + ".part.jpg"))
++        os.replace(path + ".part.jpg", path)
++
++    def _capture(self, argv):
++        proc = self.popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
++                          start_new_session=True)
++        try:
++            _, err = proc.communicate(timeout=10)
++        except subprocess.TimeoutExpired:
++            self._kill(proc)
++            raise Fault("hardware", "the screen capture did not finish in 10 s")
++        if proc.returncode != 0:
++            raise Fault("hardware", (err or b"").decode("utf-8", "replace")[:200] or "the capture tool failed")
++
++    def _notify_stream(self, seconds):
++        """Say on the desktop that the screen is live, and how to end it (config `notify`). Best effort."""
++        if not self.notify:
++            return
++        title = "Your screen is live on OpenVibe"
++        body = "Shown to your own account for %d min. Run `openvibe-node stop` to end it now." % max(1, seconds // 60)
++        argv = None
++        if self.platform.startswith("linux") and (self.env.get("DISPLAY") or self.env.get("WAYLAND_DISPLAY")) and self.which("notify-send"):
++            argv = ["notify-send", "--app-name=OpenVibe Node", "--", title, body]
++        elif self.platform == "darwin" and self.which("osascript"):
++            quote = lambda s: '"%s"' % s.replace("\\", "\\\\").replace('"', '\\"')
++            argv = ["osascript", "-e", "display notification %s with title %s" % (quote(body), quote(title))]
++        if argv is None:
++            return
++        try:
++            self.popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
++                       start_new_session=True)
++        except OSError:
++            pass
++
      @staticmethod
      def _encode(path, max_width):
          try:
ddiff --git a/plugins/computer/tests/test_computer.py b/plugins/computer/tests/test_computer.py
iindex 1588ab0..589d164 100644
---- a/plugins/computer/tests/test_computer.py
++++ b/plugins/computer/tests/test_computer.py
@@@ -418,6 +418,81 @@ class ComputerTest(unittest.TestCase):
          self.assertEqual(base64.b64decode(d["content_b64"])[:2], b"\xff\xd8")
  
  
++    # ---- the live screen -----------------------------------------------------------------------------------------
++    def boot_stream(self, level="observe", **config):
++        """A plugin whose screen tool copies one JPEG, so frames come without a display."""
++        self.boot(level, notify=False, **config)
++        src = os.path.join(self.dir, "screen.jpg")
++        with open(src, "wb") as f:
++            f.write(b"\xff\xd8\xff\xe0" + b"\x00" * 64 + b"\xff\xd9")
++        self.plugin._screen_tool = lambda: ("scrot", lambda out: [sys.executable, "-c", "import shutil,sys; shutil.copy(sys.argv[1], sys.argv[2])", src, out])
++        self.plugin._have_pillow = lambda: False
++        self.plugin._native_jpeg = lambda tool, path: [sys.executable, "-c", "import shutil,sys; shutil.copy(sys.argv[1], sys.argv[2])", src, path]
++
++    def test_screen_stream_emits_frames_at_every_level_and_stops(self):
++        self.boot_stream("observe")
++        r = reply(self.out, self.cmd("screen_stream", {"action": "start", "seconds": 60, "fps": 5}))
++        self.assertEqual(r["op"], "ack")
++        d = r["data"]
++        self.assertEqual((d["streaming"], d["seconds"], d["fps"], d["tool"]), (True, 60, 5, "scrot"))
++        frame = self.out.wait_for(lambda m: m.get("op") == "video" and m.get("stream") == "screen")
++        self.assertEqual(frame["format"], "jpeg")
++        self.assertTrue(frame["path"].startswith(tempfile.gettempdir()))
++        self.out.wait_for(lambda m: m.get("op") == "video" and m.get("path", "").endswith("frame-1.jpg"))
++        self.assertEqual(self.plugin.poll()["sensors"]["computer_screen_stream"], "live")
++        stop = reply(self.out, self.cmd("screen_stream", {"action": "stop"}))
++        self.assertEqual(stop["data"], {"streaming": False, "stopped": True})
++        ended = self.out.wait_for(lambda m: m.get("op") == "event" and m.get("name") == "screen_stream_ended")
++        self.assertEqual((ended["sid"], ended["reason"]), (d["sid"], "stopped by its owner"))
++        self.assertFalse(os.path.exists(os.path.dirname(frame["path"])), "the frames' folder is removed")
++        self.assertEqual(self.plugin.poll()["sensors"]["computer_screen_stream"], "off")
++
++    def test_screen_stream_ends_on_the_kill_switch_and_is_refused_while_latched(self):
++        self.boot_stream("trusted")
++        first = reply(self.out, self.cmd("screen_stream", {"action": "start", "seconds": 60}))["data"]
++        self.rt.dispatch({"op": "estop"})
++        ended = self.out.wait_for(lambda m: m.get("op") == "event" and m.get("name") == "screen_stream_ended")
++        self.assertEqual((ended["sid"], ended["reason"]), (first["sid"], "stopped on the machine"))
++        r = reply(self.out, self.cmd("screen_stream", {"action": "start"}))
++        self.assertEqual((r["op"], r["fault_code"]), ("nack", "local_stop"))
++
++    def test_screen_stream_replaces_and_ends_on_its_own(self):
++        self.boot_stream()
++        a = reply(self.out, self.cmd("screen_stream", {"action": "start", "seconds": 60}))["data"]
++        b = reply(self.out, self.cmd("screen_stream", {"action": "start", "seconds": 30}))["data"]
++        self.assertNotEqual(a["sid"], b["sid"])
++        ended = self.out.wait_for(lambda m: m.get("op") == "event" and m.get("name") == "screen_stream_ended")
++        self.assertEqual((ended["sid"], ended["reason"]), (a["sid"], "replaced"))
++        # Time up: a clock moved past the end stops the frames.
++        self.plugin.stream["until"] = 0
++        late = self.out.wait_for(lambda m: m.get("op") == "event" and m.get("name") == "screen_stream_ended" and m.get("sid") == b["sid"])
++        self.assertEqual(late["reason"], "time is up")
++
++    def test_screen_stream_refusals(self):
++        self.boot("observe", allow_screen_stream=False)
++        self.plugin._screen_tool = lambda: ("scrot", lambda out: ["true"])
++        r = reply(self.out, self.cmd("screen_stream", {"action": "start"}))
++        self.assertEqual(r["fault_code"], "unsupported")
++        self.boot("observe")
++        r = reply(self.out, self.cmd("screen_stream", {"action": "start"}))
++        self.assertEqual((r["fault_code"], r["message"]), ("unsupported", "no screen capture tool on this machine"))
++        self.plugin._screen_tool = lambda: ("gnome-screenshot", lambda out: ["true"])
++        self.plugin._have_pillow = lambda: False
++        r = reply(self.out, self.cmd("screen_stream", {"action": "start"}))
++        self.assertIn("install Pillow", r["message"])
++        r = reply(self.out, self.cmd("screen_stream", {"action": "watch"}))
++        self.assertEqual(r["fault_code"], "bad_value")
++
++    def test_screen_stream_capability_and_native_jpeg_tools(self):
++        c = Computer(env={}, platform="linux")
++        self.assertEqual(c._native_jpeg(("grim", None), "/t/f.jpg"), ["grim", "-t", "jpeg", "-q", "70", "/t/f.jpg"])
++        self.assertEqual(c._native_jpeg(("screencapture", None), "/t/f.jpg"), ["screencapture", "-x", "-t", "jpg", "/t/f.jpg"])
++        self.assertIsNone(c._native_jpeg(("gnome-screenshot", None), "/t/f.jpg"))
++        which = lambda name: "/usr/bin/" + name if name == "grim" else None
++        w = Computer(which=which, env={"WAYLAND_DISPLAY": "wayland-0"}, platform="linux")
++        self.assertTrue(w.describe({"level": "observe"})["screen"]["stream"])
++        self.assertFalse(w.describe({"level": "observe", "allow_screen_stream": False})["screen"]["stream"])
++
      # ---- input ---------------------------------------------------------------------------------------------------
      def boot_input(self, level, platform="linux", tool="xdotool", size=(2560, 1440)):
          self.argvs = []