eco web: task queues -- what runs and waits, scan progress, output, controls

- queues page: per queue the running task with its scan's step n/N, the
  waiting and the last finished tasks (errors), each task's output live
- controls with writes on (all items, since a queue moves whatever its
  tasks move): pause/resume the queue and the running scan, stop the
  running task, move a waiting task up/to the top, remove it, drop all
  waiting, stop everything -- nothing adds or repeats a task
- task lines name devices instead of their repr (a whole status table)
- demo: a paused queue with three simulated scans of stage.x

docs/web_app.md

@@ -76,6 +76,12 @@ python -m eco.web --demo --writes
   laptop), or use Zoom +/−, Wide and Home. *Settings* shows pan, tilt, zoom,
   iris and focus as rows. All viewers share one connection to the camera; the
   stream pauses while the page is in the background.
+- **Queues** (start page): the session's task queues ({doc}`queues`): the
+  running task with the step its scan is at, the waiting tasks, the last
+  finished ones with their errors, and each task's output, live. With writes
+  on: pause/resume the queue, pause/resume the running scan at its next step,
+  stop the running task, move a waiting task up or remove it, drop all
+  waiting, stop everything. Nothing can be added to a queue from here.
 - **Panel**: a device's schematic, refreshed every 3 s. Tapping a symbol
   opens that device; the panel's commands are not run from the web app.
 

eco/web/__main__.py

@@ -51,10 +51,13 @@ def main(argv=None):
     if args.demo:
         from eco.web.demo import DemoNamespace
 
+        from eco.web.demo import demo_queues
+
         namespace = DemoNamespace()
+        backend = SessionBackend(namespace, queues=demo_queues(namespace))
     else:
-        namespace = scope_namespace(args.scope, args.lazy)
-    gateway = WebGateway(SessionBackend(namespace), host=args.host, port=args.port,
+        backend = SessionBackend(scope_namespace(args.scope, args.lazy))
+    gateway = WebGateway(backend, host=args.host, port=args.port,
                          token=args.token, writes=args.writes).start()
     print(connection_help(gateway))
     # a terminated server must still stop the jogs its clients hold

eco/web/backends.py

@@ -40,8 +40,9 @@ def attach_ca_context():
 
 
 class SessionBackend:
-    def __init__(self, namespace, label=None):
+    def __init__(self, namespace, label=None, queues=None):
         self.namespace = namespace
+        self._queues = queues  # default: the session's eco.utilities.task_queue.queues
         self.label = label or getattr(namespace, "name", None) or "eco"
         self._init_lock = threading.Lock()
         self._index_lock = threading.Lock()
@@ -98,6 +99,14 @@ class SessionBackend:
             obj = obj.__wrapped__
         return obj
 
+    def queues(self):
+        """{name: TaskQueue} of the session (eco.utilities.task_queue.queues)."""
+        if self._queues is not None:
+            return self._queues
+        from eco.utilities import task_queue
+
+        return task_queue.queues
+
     def search_index(self):
         """[(full name, kind, object)] of every adjustable/detector in the
         initialized components, rebuilt when that set changes. Shared by all

eco/web/client_session.py

@@ -447,3 +447,40 @@ class ClientSession:
             raise ProtocolError(f"unknown ptz action {action!r}")
         logger.info("web client %s: %s ptz %s", self.client_id, name, action)
         return {"done": action}
+
+    # -- task queues: what runs and waits, and the controls for it -----------------------------
+
+    def _queue(self, name):
+        queue = dict(self.backend.queues()).get(name)  # a copy: queues[missing] would create one
+        if queue is None:
+            raise NotFound(f"no queue {name!r}")
+        return queue
+
+    def _op_queues(self):
+        """Every queue of the session (eco.web.queues.queue_json)."""
+        from eco.web import queues
+
+        return {"queues": [queues.queue_json(q) for q in list(dict(self.backend.queues()).values())],
+                "now": time.time()}
+
+    def _op_queue_output(self, queue, task, since=0):
+        """What task #`task` of `queue` printed after character `since`."""
+        from eco.web import queues
+
+        found = queues.find_task(self._queue(queue), task)
+        if found is None:
+            raise NotFound(f"no task #{task} in queue {queue!r}")
+        text, offset, clipped = queues.output_since(found, since)
+        return {"text": text, "offset": offset, "clipped": clipped, "task": queues.task_json(found)}
+
+    def _op_queue_action(self, queue, action, task=None):
+        """Pause/resume/stop/clear a queue, its running scan, or move/remove a
+        waiting task (eco.web.queues.ACTIONS) -- writes must be on. Nothing
+        here adds or starts a new task."""
+        from eco.web import queues
+
+        if self.writes is not True:  # not per item: a queue moves whatever its tasks move
+            raise WritesDisabled("queue controls need writes enabled for this web session")
+        result = queues.act(self._queue(queue), action, task)
+        logger.info("web client %s: queue %s %s %s", self.client_id, queue, action, task or "")
+        return {"done": result}

eco/web/demo.py

@@ -373,6 +373,71 @@ class Sample(Assembly):
         self._append(Cryostat, name="cryo", is_display="recursive")
 
 
+class SimScan:
+    """The parts of eco.acquisition.scan.StepScan the queue view reads:
+    steps done/to do, pause/resume at the next step, stop."""
+
+    def __init__(self, positions):
+        self._values_todo = list(positions)
+        self._values_done = []
+        self._pause_event = threading.Event()
+        self._pause_event.set()
+        self._stop_requested = False
+
+    def pause(self):
+        self._pause_event.clear()
+
+    def resume(self):
+        self._pause_event.set()
+
+    def is_paused(self):
+        return not self._pause_event.is_set()
+
+    def request_stop(self):
+        self._stop_requested = True
+        self._pause_event.set()
+
+
+def demo_scan(motor, detector, start, stop, steps, step_time=0.5):
+    """A queued 'scan' of a simulated motor: registers a SimScan with its
+    task (as StepScan does), moves step by step, prints a line per step."""
+    from eco.utilities import task_queue
+
+    positions = [start + (stop - start) * i / (steps - 1) for i in range(steps)]
+    scan = SimScan(positions)
+    task = task_queue.current_task()
+    if task is not None:
+        task.scans.append(scan)
+    print(f"scan {motor.name} {start} .. {stop}, {steps} steps")
+    while scan._values_todo:
+        while not scan._pause_event.wait(0.1):
+            pass
+        if scan._stop_requested:
+            print("stopped on request")
+            return
+        value = scan._values_todo[0]
+        motor.set_target_value(value).wait()
+        time.sleep(step_time)
+        print(f"step {len(scan._values_done) + 1:3d}/{steps}  {motor.name} = {motor.get_current_value():8.4f}"
+              f"  {detector.name} = {detector.get_current_value():9.2f}")
+        scan._values_done.append(scan._values_todo.pop(0))
+    print("scan done")
+
+
+def demo_queues(namespace):
+    """A queue "default" with three scans waiting, paused (so they wait for
+    Resume in the web client), and an empty queue "align"."""
+    from eco.utilities.task_queue import TaskQueue
+
+    stage = namespace.resolve_item("stage")
+    default = TaskQueue("default", namespace={})
+    default.pause()  # before the first append starts its worker
+    for start, stop, steps in ((-1, 1, 40), (-0.5, 0.5, 30), (0, 2, 20)):
+        default.append(demo_scan, stage.x, stage.diode, start, stop, steps, test=False,
+                       task_description=f"x {start}..{stop}")
+    return {"default": default, "align": TaskQueue("align", namespace={})}
+
+
 def _broken(name=None):
     raise ConnectionError("simulated: IOC not reachable")
 

eco/web/queues.py

@@ -0,0 +1,152 @@
+"""The session's task queues (eco.utilities.task_queue) for the web client:
+what runs, what waits, how far the running scan is, what a task printed --
+and the controls of the Qt queue panel (ScanQueuePanel) that act on what is
+already queued. Nothing here adds, repeats or edits a task.
+"""
+import re
+
+from eco.web.viewmodels import _clip
+
+HISTORY = 10  # finished tasks listed per queue
+OUTPUT_CHUNK = 20000  # characters of output per reply at most (the newest)
+
+# action -> (needs a task id, what it calls)
+ACTIONS = {
+    "pause": False,  # don't start the next task
+    "resume": False,
+    "clear": False,  # drop every waiting task
+    "stop": False,  # stop the running task and drop the waiting ones
+    "pause_running": False,  # pause the running scan at its next step
+    "resume_running": False,
+    "stop_running": False,  # its scans stop at their next step, else KeyboardInterrupt
+    "remove": True,  # drop one waiting task
+    "move_up": True,
+    "move_top": True,
+}
+
+
+def _scan(task):
+    scan = task.scan
+    if scan is None:
+        return None
+    try:
+        done = len(getattr(scan, "_values_done", []))
+        total = done + len(getattr(scan, "_values_todo", []))
+        paused = bool(scan.is_paused())
+    except Exception:
+        return None
+    return {"done": done, "total": total, "paused": paused,
+            "stopping": bool(getattr(scan, "_stop_requested", False))}
+
+
+_ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
+
+
+def _short(value):
+    """An argument as the task line shows it: an eco item by its name (its
+    repr is a whole status table), plain values by repr, anything else by
+    its type."""
+    alias = getattr(value, "alias", None)
+    if alias is not None:
+        try:
+            return alias.get_full_name()
+        except Exception:
+            pass
+    if value is None or isinstance(value, (bool, int, float, complex, str, bytes)):
+        return repr(value)
+    if isinstance(value, (list, tuple)) and len(value) <= 8:
+        inner = ", ".join(_short(v) for v in value)
+        return f"[{inner}]" if isinstance(value, list) else f"({inner}{',' if len(value) == 1 else ''})"
+    if isinstance(value, dict) and len(value) <= 8:
+        return "{" + ", ".join(f"{k!r}: {_short(v)}" for k, v in value.items()) + "}"
+    return f"<{type(value).__name__}>"
+
+
+def task_source(task):
+    """Task.source, but with short arguments (see _short)."""
+    if isinstance(task.command, str):
+        return _ANSI.sub("", task.command.strip())
+    name = getattr(task.command, "__qualname__", None) or type(task.command).__name__
+    parts = [_short(a) for a in task.args] + [f"{k}={_short(v)}" for k, v in task.kwargs.items()]
+    return f"{name}({', '.join(parts)})"
+
+
+def task_json(task):
+    out = {
+        "id": task.id,
+        "source": _clip(task_source(task)),
+        "description": _clip(str(task.description or "")),
+        "state": task.state,
+        "started": task.started,
+        "finished": task.finished,
+    }
+    if task.exception is not None:
+        out["error"] = _clip(f"{type(task.exception).__name__}: {task.exception}")
+    scan = _scan(task) if task.state == "running" else None
+    if scan is not None:
+        out["scan"] = scan
+    return out
+
+
+def queue_json(queue):
+    current = queue.current
+    return {
+        "name": queue.name,
+        "paused": queue.paused,
+        "running": task_json(current) if current is not None else None,
+        "pending": [task_json(t) for t in queue.pending],
+        "history": [task_json(t) for t in reversed(queue.history[-HISTORY:])],
+    }
+
+
+def find_task(queue, task_id):
+    return queue._find(int(task_id), everywhere=True)
+
+
+def output_since(task, since=0):
+    """(text, offset): what `task` printed after character `since`, at most
+    OUTPUT_CHUNK characters (the newest). The oldest output of a very chatty
+    task is dropped by the queue itself; an offset beyond the end starts over."""
+    text = task.output
+    since = int(since or 0)
+    if since > len(text):
+        since = 0
+    chunk = text[since:]
+    clipped = len(chunk) > OUTPUT_CHUNK
+    if clipped:
+        chunk = chunk[-OUTPUT_CHUNK:]
+    return chunk, len(text), clipped
+
+
+def act(queue, action, task_id=None):
+    """Run one allowed action on `queue`; returns a short description."""
+    if action not in ACTIONS:
+        raise ValueError(f"unknown queue action {action!r}")
+    if ACTIONS[action]:
+        if task_id is None:
+            raise ValueError(f"{action} needs a task")
+        task = queue._find(int(task_id))
+        if task is None:
+            raise ValueError(f"task #{task_id} is not waiting in queue {queue.name!r} (any more)")
+        if action == "remove":
+            queue.remove(task)
+        else:
+            pending = queue.pending
+            index = 0 if action == "move_top" else max(0, pending.index(task) - 1)
+            queue.move(task, index)
+        return f"task #{task.id}: {action}"
+    if action == "pause":
+        queue.pause()
+    elif action == "resume":
+        queue.resume()
+    elif action == "clear":
+        return f"dropped {len(queue.clear())} waiting tasks"
+    elif action == "stop":
+        dropped = queue.stop()
+        return f"stopping the running task, dropped {len(dropped)} waiting"
+    elif action in ("pause_running", "resume_running", "stop_running"):
+        method = {"pause_running": queue.pause_running_task, "resume_running": queue.resume_running_task,
+                  "stop_running": queue.stop_running_task}[action]
+        if method() is None:
+            raise ValueError("nothing is running" if action == "stop_running" else "the running task has no scan to pause")
+    return action

eco/web/static/app.js

@@ -293,6 +293,8 @@ function route() {
   else if (hash.startsWith("/tweak/")) showTweak(hash.slice(7));
   else if (hash.startsWith("/jog/")) showJog(hash.slice(5));
   else if (hash.startsWith("/cam/")) showCamera(hash.slice(5));
+  else if (hash === "/queues") showQueues();
+  else if (hash.startsWith("/qout/")) showTaskOutput(hash.slice(6));
   else showHome();
 }
 
@@ -342,7 +344,9 @@ async function showHome() {
     autocapitalize: "off", autocomplete: "off", spellcheck: "false",
     oninput: (e) => { S.filter = e.target.value; render(); } });
   render();
-  setMain(search, sections);
+  const tools = el("div", { class: "toolbar" },
+    el("a", { class: "btn link-btn", href: "#/queues", text: "Queues" }));
+  setMain(tools, search, sections);
 }
 
 // ---- device: an assembly or a single item -----------------------------------------------------
@@ -1127,6 +1131,149 @@ async function showCamera(name) {
   if (document.visibilityState === "visible") start();
 }
 
+// ---- task queues: what runs and waits, pause/resume/stop -----------------------------------------------
+
+const QUEUE_POLL = 1000; // ms
+
+function duration(seconds) {
+  if (seconds === null || seconds === undefined) return "";
+  const s = Math.max(0, Math.round(seconds));
+  return s < 60 ? `${s} s` : s < 3600 ? `${Math.floor(s / 60)} min ${s % 60} s` : `${Math.floor(s / 3600)} h ${Math.floor((s % 3600) / 60)} min`;
+}
+
+async function showQueues() {
+  const view = newView("queues", "");
+  setTitle("Queues", true);
+  const box = el("div", { class: "queues" }, el("div", { class: "loading", text: "loading queues…" }));
+  setMain(box, el("p", { class: "doc", text: "The session's task queues (%queue, scans with queue=…). " +
+    "Controls act on what is already queued; nothing is added here." }));
+  const writes = !!S.hello.writes;
+  const act = (queue, action, task, question) => {
+    const go = () => request("queue_action", { queue, action, task }).then(load, (e) => toast(e.message, true));
+    question ? confirmDialog(question, "", go) : go();
+  };
+  const button = (text, onclick, cls = "") =>
+    el("button", { class: `btn small ${cls}`, text, onclick, disabled: !writes, title: writes ? "" : "writes are off in this web session" });
+
+  const taskLine = (q, t, kind) => {
+    const took = t.started ? duration((t.finished || serverNow()) - t.started) : "";
+    const head = el("div", { class: "task-head" },
+      el("span", { class: `task-state ${t.state}`, text: t.state }),
+      el("span", { class: "task-id", text: `#${t.id}` }),
+      t.description ? el("span", { class: "task-desc", text: t.description }) : null,
+      took ? el("span", { class: "task-time", text: took }) : null);
+    const node = el("div", { class: `task ${kind}` }, head, el("div", { class: "task-source", text: t.source }));
+    if (t.error) node.append(el("div", { class: "task-error", text: t.error }));
+    if (t.scan) {
+      const pct = t.scan.total ? (100 * t.scan.done) / t.scan.total : 0;
+      node.append(el("div", { class: "progress" }, el("div", { class: "bar", style: `width:${pct.toFixed(1)}%` })),
+        el("div", { class: "task-scan", text: `step ${t.scan.done} / ${t.scan.total}` +
+          (t.scan.stopping ? " · stopping" : t.scan.paused ? " · scan paused" : "") }));
+    }
+    const tools = el("div", { class: "task-tools" });
+    if (kind !== "pending") tools.append(el("a", { class: "btn small link-btn", href: `#/qout/${encodeURIComponent(q.name)}/${t.id}`, text: "Output" }));
+    if (kind === "running") {
+      if (t.scan) tools.append(t.scan.paused ? button("Resume scan", () => act(q.name, "resume_running"))
+        : button("Pause scan", () => act(q.name, "pause_running")));
+      tools.append(button("Stop task", () => act(q.name, "stop_running", undefined,
+        `Stop task #${t.id}? Its scan stops at the next step; the queue pauses.`), "stop"));
+    } else if (kind === "pending") {
+      tools.append(button("Top", () => act(q.name, "move_top", t.id)), button("↑", () => act(q.name, "move_up", t.id)),
+        button("Remove", () => act(q.name, "remove", t.id, `Remove task #${t.id} from queue ${q.name}?`), "stop"));
+    }
+    if (tools.childElementCount) node.append(tools);
+    return node;
+  };
+
+  const render = (data) => {
+    if (!data.queues.length) {
+      box.replaceChildren(el("div", { class: "empty", text: "No queues in this session yet." }));
+      return;
+    }
+    box.replaceChildren(...data.queues.map((q) => {
+      const state = q.running ? (q.paused ? "running, then paused" : "running") : q.paused ? "paused" : q.pending.length ? "starting" : "idle";
+      const card = el("section", { class: "queue-card" },
+        el("div", { class: "queue-head" },
+          el("h2", { text: q.name }), el("span", { class: `badge q-${q.paused ? "paused" : q.running ? "running" : "idle"}`, text: state }),
+          q.paused ? button("Resume queue", () => act(q.name, "resume", undefined, q.pending.length ? `Resume queue ${q.name}? ${q.pending.length} waiting task(s) will start.` : null))
+            : button("Pause queue", () => act(q.name, "pause"))));
+      if (q.running) card.append(taskLine(q, q.running, "running"));
+      if (q.pending.length) {
+        card.append(el("div", { class: "section-title", text: `Waiting (${q.pending.length})` }),
+          ...q.pending.map((t) => taskLine(q, t, "pending")),
+          el("div", { class: "task-tools" },
+            button("Drop all waiting", () => act(q.name, "clear", undefined, `Drop all ${q.pending.length} waiting tasks of ${q.name}?`), "stop")));
+      }
+      if (q.running || q.pending.length) {
+        card.append(el("div", { class: "task-tools" }, button("Stop everything", () => act(q.name, "stop", undefined,
+          `Stop the running task of ${q.name} and drop everything waiting?`), "stop")));
+      }
+      if (q.history.length) {
+        const list = el("div", {}, ...q.history.map((t) => taskLine(q, t, "done")));
+        card.append(el("details", { class: "hidden-items" }, el("summary", { text: `Finished (${q.history.length})` }), list));
+      }
+      return card;
+    }));
+  };
+  let busy = false;
+  const open = new Set(); // keep "Finished" folds open across refreshes
+  const load = async () => {
+    if (busy || S.view !== view || document.visibilityState !== "visible") return;
+    busy = true;
+    try {
+      box.querySelectorAll("section.queue-card").forEach((c) => {
+        if (c.querySelector("details")?.open) open.add(c.querySelector("h2").textContent);
+        else open.delete(c.querySelector("h2")?.textContent);
+      });
+      const data = await request("queues");
+      if (S.view !== view) return;
+      render(data);
+      box.querySelectorAll("section.queue-card").forEach((c) => {
+        const d = c.querySelector("details");
+        if (d && open.has(c.querySelector("h2").textContent)) d.open = true;
+      });
+    } catch (e) {
+      if (S.view === view) box.replaceChildren(el("div", { class: "empty", text: `could not read the queues: ${e.message}` }));
+    } finally {
+      busy = false;
+    }
+  };
+  await load();
+  view.timers.push(setInterval(load, QUEUE_POLL));
+}
+
+async function showTaskOutput(path) {
+  const slash = path.lastIndexOf("/");
+  const queue = path.slice(0, slash), id = parseInt(path.slice(slash + 1), 10);
+  const view = newView("qout", path);
+  setTitle(`${queue} #${id}`, true);
+  const head = el("div", { class: "task-head" });
+  const pre = el("pre", { class: "task-output", text: "" });
+  const follow = el("label", { class: "follow" }, el("input", { type: "checkbox", checked: true }), " follow");
+  setMain(head, pre, follow);
+  let offset = 0, busy = false;
+  const load = async () => {
+    if (busy || S.view !== view || document.visibilityState !== "visible") return;
+    busy = true;
+    try {
+      const r = await request("queue_output", { queue, task: id, since: offset });
+      if (S.view !== view) return;
+      if (r.offset < offset || r.clipped) pre.textContent = r.clipped ? "… (older output not shown)\n" : "";
+      offset = r.offset;
+      if (r.text) pre.append(r.text);
+      head.replaceChildren(el("span", { class: `task-state ${r.task.state}`, text: r.task.state }),
+        el("span", { class: "task-source", text: r.task.source }));
+      if (follow.querySelector("input").checked && r.text) pre.scrollTop = pre.scrollHeight;
+    } catch (e) {
+      if (S.view === view) head.replaceChildren(el("span", { class: "task-error", text: e.message }));
+    } finally {
+      busy = false;
+    }
+  };
+  await load();
+  view.timers.push(setInterval(load, QUEUE_POLL));
+}
+
 // ---- SVG panels ------------------------------------------------------------------------------------
 
 const NAME_RE = /^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$/;

eco/web/static/style.css

@@ -189,3 +189,29 @@ a.btn:not(.primary) { color: var(--link); }
 .cam-box { display: none; position: absolute; border: 2px dashed #fff; background: rgba(255,255,255,.12); pointer-events: none; }
 .cam-tools { margin-top: 8px; }
 .btn.on { background: var(--busy); color: #000; border-color: var(--busy); }
+
+/* task queues */
+.queue-card { background: var(--surface); border: 1px solid var(--line); border-radius: 12px; padding: 10px 12px; margin-bottom: 12px; }
+.queue-head { display: flex; align-items: center; gap: 8px; flex-wrap: wrap; }
+.queue-head h2 { margin: 0; font-size: 18px; flex: 1; }
+.badge.q-running { background: var(--busy); color: #000; }
+.badge.q-paused { background: var(--err); color: #fff; }
+.badge.q-idle { background: var(--line); color: var(--text); }
+.task { border-top: 1px solid var(--line); padding: 8px 0; }
+.task-head { display: flex; flex-wrap: wrap; align-items: baseline; gap: 4px 10px; }
+.task-state { font-size: 12px; font-weight: 700; text-transform: uppercase; color: var(--muted); }
+.task-state.running { color: var(--busy); }
+.task-state.done { color: var(--ok); }
+.task-state.error { color: var(--err); }
+.task-id { color: var(--muted); font-size: 13px; }
+.task-desc { font-weight: 600; }
+.task-time { color: var(--muted); font-size: 13px; margin-left: auto; }
+.task-source { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; font-size: 13px; overflow-wrap: anywhere; }
+.task-error { color: var(--err); font-size: 13px; overflow-wrap: anywhere; }
+.task-scan { color: var(--muted); font-size: 13px; }
+.task-tools { display: flex; flex-wrap: wrap; gap: 6px; margin-top: 6px; }
+.progress { height: 8px; border-radius: 4px; background: var(--line); overflow: hidden; margin-top: 6px; }
+.progress .bar { height: 100%; background: var(--busy); }
+.task-output { background: var(--surface); border: 1px solid var(--line); border-radius: 12px; padding: 10px; max-height: 65vh; overflow: auto;
+  font-size: 12px; white-space: pre-wrap; overflow-wrap: anywhere; }
+.follow { display: inline-flex; gap: 6px; align-items: center; color: var(--muted); font-size: 14px; }

eco/web/static/sw.js

@@ -3,7 +3,7 @@
 // SVG panels or the websocket -- a stale beamline value shown as current
 // would be worse than none. Network first, so a new client version is
 // picked up as soon as the gateway is reachable.
-const CACHE = "eco-shell-v4";
+const CACHE = "eco-shell-v5";
 const SHELL = ["/", "/static/app.js", "/static/strip.js", "/static/style.css", "/icon.svg", "/manifest.webmanifest"];
 
 self.addEventListener("install", (event) => {

tests/test_web_client_session.py

@@ -326,3 +326,102 @@ def test_ptz_needs_writes_and_passes_the_access_gate(client, writer, monkeypatch
     writer.call("describe", name="cam_demo")
     reply = writer.call("ptz", name="cam_demo", action="home")
     assert not reply["ok"] and "not your camera" in reply["error"]["message"]
+
+
+# -- task queues -------------------------------------------------------------------------
+
+
+def _queue_client(writes):
+    from eco.web.demo import demo_queues
+
+    ns = DemoNamespace()
+    c = Client(writes=writes)
+    c.session.close()
+    queues = demo_queues(ns)
+    from eco.web.client_session import ClientSession
+
+    c.session = ClientSession(SessionBackend(ns, queues=queues), c.messages.put, writes=writes, poll_interval=0.05)
+    return c, queues
+
+
+def _wait(predicate, seconds=5.0):
+    deadline = time.monotonic() + seconds
+    while not predicate() and time.monotonic() < deadline:
+        time.sleep(0.05)
+    return predicate()
+
+
+def test_queues_list_running_waiting_and_scan_progress():
+    c, queues = _queue_client(writes=True)
+    try:
+        data = c.call("queues")["data"]["queues"]
+        default = next(q for q in data if q["name"] == "default")
+        assert default["paused"] and default["running"] is None and len(default["pending"]) == 3
+        assert default["pending"][0]["description"] == "x -1..1"
+        assert c.call("queue_action", queue="default", action="resume")["ok"]
+        assert _wait(lambda: (queues["default"].current is not None and queues["default"].current.scan is not None
+                              and queues["default"].current.scan._values_done))
+        running = next(q for q in c.call("queues")["data"]["queues"] if q["name"] == "default")["running"]
+        assert running["state"] == "running" and running["scan"]["total"] == 40 and running["scan"]["done"] >= 1
+        # pause the scan, then stop the task: the queue pauses, the rest waits
+        assert c.call("queue_action", queue="default", action="pause_running")["ok"]
+        assert _wait(lambda: queues["default"].current.scan.is_paused())
+        assert c.call("queue_action", queue="default", action="stop_running")["ok"]
+        assert _wait(lambda: queues["default"].current is None)
+        q = next(q for q in c.call("queues")["data"]["queues"] if q["name"] == "default")
+        assert q["paused"] and len(q["pending"]) == 2 and q["history"][0]["state"] == "done"
+        out = c.call("queue_output", queue="default", task=q["history"][0]["id"])["data"]
+        assert "stopped on request" in out["text"] and out["offset"] == len(out["text"])
+        again = c.call("queue_output", queue="default", task=q["history"][0]["id"], since=out["offset"])["data"]
+        assert again["text"] == ""
+    finally:
+        for queue in queues.values():
+            queue.stop()
+            queue.close()
+        c.session.close()
+
+
+def test_waiting_tasks_can_be_moved_and_removed():
+    c, queues = _queue_client(writes=True)
+    try:
+        ids = [t.id for t in queues["default"].pending]
+        assert c.call("queue_action", queue="default", action="move_top", task=ids[2])["ok"]
+        assert [t.id for t in queues["default"].pending] == [ids[2], ids[0], ids[1]]
+        assert c.call("queue_action", queue="default", action="move_up", task=ids[1])["ok"]
+        assert [t.id for t in queues["default"].pending] == [ids[2], ids[1], ids[0]]
+        assert c.call("queue_action", queue="default", action="remove", task=ids[1])["ok"]
+        assert [t.id for t in queues["default"].pending] == [ids[2], ids[0]]
+        assert c.call("queue_action", queue="default", action="clear")["ok"]
+        assert queues["default"].pending == []
+        reply = c.call("queue_action", queue="default", action="remove", task=ids[0])
+        assert not reply["ok"] and "not waiting" in reply["error"]["message"]
+        assert not c.call("queue_action", queue="default", action="append")["ok"]
+        assert not c.call("queue_action", queue="nope", action="pause")["ok"]
+        assert "nope" not in queues
+    finally:
+        for queue in queues.values():
+            queue.close()
+        c.session.close()
+
+
+def test_queue_controls_need_writes():
+    c, queues = _queue_client(writes=False)
+    try:
+        assert c.call("queues")["ok"]
+        reply = c.call("queue_action", queue="default", action="resume")
+        assert not reply["ok"] and reply["error"]["type"] == "WritesDisabled"
+        assert queues["default"].paused
+    finally:
+        for queue in queues.values():
+            queue.close()
+        c.session.close()
+
+
+def test_task_lines_name_devices_instead_of_their_status_tables():
+    from eco.utilities.task_queue import Task
+    from eco.web.queues import task_source
+
+    ns = DemoNamespace()
+    stage = ns.resolve_item("stage")
+    task = Task(lambda *a, **k: None, (stage.x, -1, [1, 2]), {"counters": [stage.diode], "step_info": None})
+    assert task_source(task) == "test_task_lines_name_devices_instead_of_their_status_tables.<locals>.<lambda>(stage.x, -1, [1, 2], counters=[stage.diode], step_info=None)"

web_frontend/PARITY.md

@@ -17,7 +17,9 @@ opened as a view; SVG panels; details (the grids' ⋯ menu); pinned live strip
 charts with running average; a picker over the initialized components; a
 tweak view (one axis); a **jog** view for hardware jogs (motor records:
 hold to move, jog speed ÷2/×2/set, dead-man stop in the gateway); a live view
-of MJPEG PTZ network cameras (cam_west & co.) with tap-to-center/drag-to-zoom. Started
+of MJPEG PTZ network cameras (cam_west & co.) with tap-to-center/drag-to-zoom;
+the task queues with scan progress, output and the queue panel's controls
+(not the launcher). Started
 with `eco web -s <scope>` or `eco.start_web()`.
 
 ## Tier 1 — core, port first

web_frontend/PROTOCOL.md

@@ -49,9 +49,9 @@ The client never sends expressions.
 | — | — | a camera's live stream is plain HTTP: `GET /api/mjpeg?name=…&quality=low|medium|full` returns `multipart/x-mixed-replace` JPEGs for an `<img>`; `describe` of such an item has `camera: {stream: "mjpeg", ptz, rotate, qualities}` |
 | `jog_speed` | `{name, value}` | sets the jog speed of `name` (its `jog_speed` item); allowed whenever writing `name` is |
 | — | — | SVG panels are plain HTTP: `GET /api/svg?name=…` returns the SVG, header `X-Eco-Prefix` is what click markers are relative to; the client re-fetches every 3 s |
-| `queues` *planned* | `{}` | queue list with tasks (mode B) |
-| `queue_call` *planned* | `{queue, action: "pause"|"resume"|"abort_current"|…}` | allow-listed subset of `TaskQueue` methods |
-| `queue_follow` *planned* | `{queue, from_task?}` | backlog + live `ev: "queue_output"` |
+| `queues` | `{}` | `{queues: [{name, paused, running: task?, pending: [task], history: [task] (newest first, 10)}], now}`; task = `{id, source, description, state, started, finished, error?, scan?: {done, total, paused, stopping}}` — the client polls it every second |
+| `queue_action` | `{queue, action, task?}` | (writes on, all items) `pause`, `resume`, `clear`, `stop`, `pause_running`, `resume_running`, `stop_running`, and for a waiting task `remove`, `move_up`, `move_top` (`eco.web.queues.ACTIONS`); nothing adds or repeats a task |
+| `queue_output` | `{queue, task, since}` | `{text, offset, clipped, task}` — what the task printed after character `since` (at most the newest 20 000 characters) |
 | `tweak` *planned* | `{session_id, action, args}` | `TweakController` methods by allow-list |
 
 ## Events

web_frontend/README.md

@@ -46,6 +46,9 @@ phone size):
   (30 s .. 1 h) with optional running average, survive a reload
 - picker: find any readable item of the initialized components by name
 - tweak view: big step buttons, ×2/÷2, go to, back to start, stop, arrow keys
+- task queues: running task with scan progress, waiting and finished tasks,
+  live task output; pause/resume queue and running scan, stop, move/remove a
+  waiting task, drop all (`eco/web/queues.py`; nothing adds or repeats a task)
 - live view of MJPEG network cameras (AxisPTZ: cam_west, ...), relayed by the
   gateway (`eco/web/camera.py`: one camera connection per quality, newest frame
   per client), with tap-to-center, drag-to-zoom, zoom steps, wide, home. Tested
@@ -54,7 +57,7 @@ phone size):
 - token auth + HttpOnly cookie, binds 127.0.0.1, ping/reconnect after network
   drops, clean shutdown, service worker (app shell only), light/dark
 
-Not yet: queues/scan progress, archiver history, memories, cam_server cameras, alarms,
+Not yet: scan launcher (adding tasks), archiver history, memories, cam_server cameras, alarms,
 status-server (shared, read-only) mode. See [PARITY.md](PARITY.md) for the order.
 
 Points to keep in mind (found while wiring it in):