hub: incremental polling (since= skeleton delta, nt rewind marker, seg ?off tail fetch) to cut relay traffic

frontends/hub.html

@@ -138,7 +138,9 @@ async function sync(first){
   if (!cur) return;
   if (!first && document.hidden) return;          // background tab costs nothing
   const name = cur, m = $('#main');
-  let q = `/api/${encodeURIComponent(name)}/messages?last=1`;
+  const olds = first ? [] : [...m.querySelectorAll('.task')];
+  const since = Math.max(0, olds.length - 1);     // rows below the tail are immutable: don't resend them
+  let q = `/api/${encodeURIComponent(name)}/messages?since=${since}`;
   if (!first && sig) q += `&sig=${encodeURIComponent(sig)}`;
   let r;
   try { r = await api(q); } catch(e){ return; }
@@ -148,19 +150,20 @@ async function sync(first){
   if (r.error){ if (first) m.innerHTML = `<div class="empty">${esc(r.error)}</div>`; return; }
 
   const tasks = r.tasks || [];
+  const total = 'nt' in r ? r.nt : tasks.length;  // an old host ignores `since` and sends everything
+  const t0 = tasks.find(t => t.i === 0);
+  // /clear (or trim) rewinds indices -> nothing to reuse, refetch from scratch
+  if (!first && (olds.length > total || (olds.length && t0 && olds[0]._input !== t0.input)))
+    return open_(name);
   const atEnd = m.scrollHeight - m.scrollTop - m.clientHeight < 80;
-  const olds = [...m.querySelectorAll('.task')];
-  // /clear (or trim) rewinds indices -> nothing to reuse, start over
-  const reset = first || olds.length > tasks.length ||
-                (olds.length && olds[0]._input !== (tasks[0] || {}).input);
-  if (reset) m.innerHTML = '<div class="hd"></div>';
+  if (first || !m.querySelector('.hd')) m.innerHTML = '<div class="hd"></div>';
   m.querySelector('.hd').textContent = r.title || name;
 
   for (const t of tasks){
     const el = m.querySelector(`.task[data-i="${t.i}"]`);
     if (el) el.update(t); else m.appendChild(taskEl(name, t));
   }
-  if (reset || atEnd) m.scrollTop = m.scrollHeight;
+  if (first || atEnd) m.scrollTop = m.scrollHeight;
 }
 
 // one agent task: input bubble + collapsible output steps
@@ -198,15 +201,20 @@ function segEl(name, i, s){
       <span class="tt"></span><span class="n"></span></summary>
       <div class="body dim">…</div>`;
   const tt = det.querySelector('.tt'), nn = det.querySelector('.n'), body = det.querySelector('.body');
-  let n = -1, loaded = -1, busy = false;           // loaded = length of the text on screen
+  let n = -1, loaded = -1, busy = false, raw = ''; // raw = full text on screen, kept so growth only fetches the tail
 
-  const pull = async () => {                       // fetch full text of this step, on demand
+  const pull = async () => {                       // fetch only what we don't have yet
     if (busy || loaded === n) return;
     busy = true;
     try {
-      const r = await api(`/api/${encodeURIComponent(name)}/seg/${i}/${s.j}`);
+      const base = `/api/${encodeURIComponent(name)}/seg/${i}/${s.j}`;
+      let off = raw.length;
+      let r = await api(`${base}?off=${off}`);
+      if (!r.error && r.n < off){ off = 0; r = await api(base); }   // step rewound (trim/clear) -> full refetch
       body.classList.remove('dim');
-      body.innerHTML = r.error ? '⚠ ' + esc(r.error) : md(r.content);
+      if (r.error){ body.innerHTML = '⚠ ' + esc(r.error); loaded = n; return; }
+      raw = (off ? raw : '') + (r.content || '');
+      body.innerHTML = md(raw);
       loaded = n;
     } catch(e){} finally { busy = false; }
   };

frontends/hub.py

@@ -1,6 +1,6 @@
 """GA Hub: `import hub` -> client (silent if no server); `python hub.py` -> WS server.
     hub.connect(agent, 'stapp')     # or override any hook: connect(a, n, put_task=, get_outputs=, abort=)
-    default put -> 'busy' if agent.is_running, else parks {'text','q'} in agent._hub_inbox for the UI
+    default put -> 'busy' if agent.is_running, else parks plain text in agent._hub_inbox for the UI
 HTTP (errors = {'error','code'} + status: offline/gone 404, busy 409, timeout 504, nosupport 501, badop 400):
     GET peers -> [{name,title,n_msgs,sig}] | {name}/messages?detail=1&sig= -> {title,tasks:[{i,input,steps:
     [{j,title,n}]}],sig} or {same:1,sig} | {name}/seg/{i}/{j}?off=N -> {content,off,n} (step bodies, tailable)
@@ -57,10 +57,12 @@ class HubClient:
         w = sum(2 if ord(ch) > 0x2E80 else 1 for ch in title)   # display width: a CJK char is worth two
         if len(ins) > 1 and w < TITLE_MIN: title = ins[-2].split('\n')[0] + ' <- ' + title
         det, rows = int(c.get('detail', 1)), []
+        since = max(0, int(c.get('since') or 0))   # delta: the client already holds rows < since (they are immutable)
         for i, (t, o) in enumerate(zip(tasks, outs)):
+            if i < since: continue
             steps = [{'j': j, 'title': self._stitle(i, j, s), 'n': len(s or '')} for j, s in enumerate(o)]
             rows.append({'i': i, 'input': t.get('input', ''), **({'steps': steps} if det else {'ns': len(o)})})
-        return {'title': title[:80], 'tasks': rows, 'sig': sig, **self.state()}
+        return {'title': title[:80], 'tasks': rows, 'nt': len(tasks), 'sig': sig, **self.state()}   # nt lets a delta client spot /clear
     def _seg(self, c):
         tasks, off = list(self.get_outputs() or []), max(0, int(c.get('off') or 0))
         i, j = int(c.get('i', -1)), int(c.get('j', -1))
@@ -92,11 +94,12 @@ def serve():
 def connect(agent, name=None, put_task=None, get_outputs=None, abort=None, fold=None):
     """One line to wire a GA host: `hub.connect(agent, 'stapp')`; any hook can still be overridden.
     Default put refuses while the agent is busy (a remote must not cut in line), else it parks
-    (text, queue) in agent._hub_inbox so a UI can pop it and echo the task as its own bubble.
+    plain text in agent._hub_inbox; the UI feeds it through its own input entrance when idle,
+    so remote tasks behave exactly like typed ones (bubble/stream/stop). Requires a live UI tab.
     Never raises: a broken hub must not break its host."""
     def _put(text):
         if getattr(agent, 'is_running', False): return {'error': f'peer {name} is busy', 'code': 'busy'}
-        agent._hub_inbox.append({'text': text, 'q': agent.put_task(text, source='hub')})
+        agent._hub_inbox.append(text)   # no put_task here: the UI's unified entrance sends it
     try:
         try: serve()                                   # best effort: bring up a local hub if none is listening
         except Exception: pass
@@ -179,7 +182,7 @@ if __name__ == '__main__':
         for n in [x for x in pcache if x not in peers]: pcache.pop(n, None)
         return rows
     @app.get('/api/{name}/messages')
-    async def api_messages(name: str, detail: int = 1, sig: str = None): return out(await ask(name, {'op': 'get', 'detail': detail, 'sig': sig}))
+    async def api_messages(name: str, detail: int = 1, sig: str = None, since: int = 0): return out(await ask(name, {'op': 'get', 'detail': detail, 'sig': sig, 'since': since}))
     @app.get('/api/{name}/seg/{i}/{j}')
     async def api_seg(name: str, i: int, j: int, off: int = 0): return out(await ask(name, {'op': 'seg', 'i': i, 'j': j, 'off': off}))
     @app.post('/api/{name}/put')