Release: Live Stream Anchor side-effects projection (#6204, @franksong2702) (#6348)

* fix: preserve anchor-owned side effects

* test: prove invisible anchor outcomes do not repaint

* docs(changelog): stamp #6204 Anchor side-effects projection

---------

Co-authored-by: Frank Song <franksong2702@gmail.com>
Co-authored-by: nesquena-hermes <agent@nesquena-hermes>

CHANGELOG.md

@@ -17,6 +17,8 @@
 
 ### Fixed
 
+- **Live Stream: a settled Worklog now shows the side effects a turn performed (memory/skill saves).** The `state_saved` SSE event (memory + skill saves) was classified as an Assistant-Turn-Anchor side effect but the renderer-neutral scene projection discarded it, so the browser only flashed a toast and a reloaded Worklog lost that record. Anchor `artifacts`/`side_effects` are now projected into the settled `activity_scene_v1`. Purely additive — turns with no side effects render byte-identical (side-effect-only turns still don't force a Worklog), and the change is compatible with the stable-run-identity plumbing. Thanks @franksong2702. (#6204, #3400)
+
 - **Kanban board columns are now scrollable on mobile.** On a phone, a touch-drag inside a Kanban column was trapped in the column's nested scroller, so you couldn't scroll the board. The mobile layout now lets the column overscroll chain to the page/section, restoring natural scrolling; desktop and the pull-to-refresh behavior are unchanged. Thanks @jpalazz2. (#6306)
 
 - **Live Stream: Assistant Turn Anchors now carry a stable run identity through the settled scene.** The activity-scene projection now threads a stable `run_id` (independent of the transport `stream_id`) through Anchor scenes and client hydration, so a settled/reloaded Worklog keeps a consistent identity and can't mis-own or duplicate rows if a producer ever emits `run_id != stream_id`. Behavior-preserving for current writers (which still emit `run_id == stream_id`) — verified byte-identical live/settle/reload render, with malformed/conflicting identities falling back to the transport identity. Thanks @franksong2702. (#6201, #3400)

docs/architecture/stable-assistant-turn-anchor-phase0.md

@@ -54,6 +54,13 @@ when that slice landed; they do not override the current coverage summary above.
   `_anchor_activity_scene` snapshot so the folded activity summary can appear
   above the final answer. This handoff does not claim Transparent Stream wiring
   or durable hard-reload scene persistence.
+- The first RFC Phase 6 ownership slice routes the production `state_saved` SSE
+  into the active Assistant Turn Anchor and carries `artifacts` and
+  `side_effects` through `activity_scene_v1` settlement persistence and reload
+  hydration. These owned outcomes remain off the readable Worklog by default;
+  memory/skill saves still surface through their existing toast and natural UI.
+  Production artifact-reference derivation and run-journal reconstruction of
+  side effects remain follow-up work.
 
 ## State Layers
 

static/assistant_turn_anchors.js

@@ -1030,6 +1030,8 @@
         final_message_ref:null,
         terminal_state:null,
         activity_rows:Object.freeze([]),
+        artifacts:Object.freeze([]),
+        side_effects:Object.freeze([]),
       });
     }
     const rows=(Array.isArray(anchor.activity_events)?anchor.activity_events:[])
@@ -1045,6 +1047,14 @@
       final_message_ref:typeof content.final_message_ref==='string'?content.final_message_ref:null,
       terminal_state:_cleanString(_own(lifecycle,'terminal_state'))||null,
       activity_rows:Object.freeze(rows),
+      artifacts:Object.freeze((Array.isArray(anchor.artifacts)?anchor.artifacts:[]).map(event=>Object.freeze({
+        ..._copyObject(event),
+        payload:Object.freeze(_copyObject(_own(event,'payload'))),
+      }))),
+      side_effects:Object.freeze((Array.isArray(anchor.side_effects)?anchor.side_effects:[]).map(event=>Object.freeze({
+        ..._copyObject(event),
+        payload:Object.freeze(_copyObject(_own(event,'payload'))),
+      }))),
     });
   }
 

static/messages.js

@@ -2649,7 +2649,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
   _scheduleAnchorRegistryCleanup(600000);
   // Applying an event and painting it are separate outcomes. Reasoning uses the
   // optional holder to decide whether a temporary visible fallback is needed.
-  function _applyToAnchor(sourceEventType, rawEventData, sseEvent, renderOutcome){
+  function _applyToAnchor(sourceEventType, rawEventData, sseEvent, renderOutcome, options={}){
     if(renderOutcome&&typeof renderOutcome==='object') renderOutcome.rendered=false;
     if(!_anchorRegistry||!_anchorApi||typeof _anchorApi.applyAssistantTurnAnchorSourceEvent!=='function') return null;
     const raw=(rawEventData&&typeof rawEventData==='object')?rawEventData:{};
@@ -2674,7 +2674,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
         sourceEvent,
         {session_id:activeSid,stream_id:streamId}
       );
-      const rendered=_renderAnchorLiveScene();
+      const rendered=options&&options.render===false?false:_renderAnchorLiveScene();
       if(renderOutcome&&typeof renderOutcome==='object') renderOutcome.rendered=rendered;
       return result;
     }catch(err){
@@ -3694,6 +3694,12 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
     }
     return false;
   }
+  function _anchorSceneHasOwnedOutcomes(scene){
+    return !!(
+      (Array.isArray(scene&&scene.artifacts)&&scene.artifacts.length)
+      || (Array.isArray(scene&&scene.side_effects)&&scene.side_effects.length)
+    );
+  }
   function _attachProjectedAnchorSceneToLastAssistant(messages){
     if(!_anchorRegistry||!Array.isArray(messages)) return false;
     let lastAsst=null;
@@ -3709,9 +3715,10 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
     if(!lastAsst) return false;
     const projectedScene=_projectLiveAnchorActivityScene();
     const scene=_completeSettledAnchorSceneForTurn(messages,lastAsstIndex,projectedScene);
-    if(scene&&Array.isArray(scene.activity_rows)&&scene.activity_rows.length){
+    const hasOwnedOutcomes=_anchorSceneHasOwnedOutcomes(scene);
+    if(scene&&Array.isArray(scene.activity_rows)&&(scene.activity_rows.length||hasOwnedOutcomes)){
       const hasWorklogRows=_anchorSceneHasWorklogWorthyRows(scene);
-      const shouldPersistScene=hasWorklogRows||scene.mode==='hide_all_activity';
+      const shouldPersistScene=hasWorklogRows||scene.mode==='hide_all_activity'||hasOwnedOutcomes;
       if(!shouldPersistScene) return false;
       lastAsst._anchor_stream_id=streamId;
       lastAsst._anchor_activity_scene=scene;
@@ -5586,6 +5593,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
       try{ d=JSON.parse(e.data||'{}'); }catch(_){}
       if((d.session_id||activeSid)!==activeSid) return;
       if(!S.session||S.session.session_id!==activeSid) return;
+      _applyToAnchor('state_saved',d,e,null,{render:false});
       _showPersistentStateToast(d.kind, d.name||'', {created:String(d.action||'').toLowerCase()==='created'});
     });
 

tests/test_issue3820_chat_activity_display_mode.py

@@ -322,7 +322,11 @@ def test_chat_activity_display_mode_settled_hide_all_scene_persists_without_work
     return_index = block.index("return hasWorklogRows;")
 
     assert "const hasWorklogRows=_anchorSceneHasWorklogWorthyRows(scene);" in block
-    assert "const shouldPersistScene=hasWorklogRows||scene.mode==='hide_all_activity';" in block
+    assert "const hasOwnedOutcomes=_anchorSceneHasOwnedOutcomes(scene);" in block
+    assert (
+        "const shouldPersistScene=hasWorklogRows||scene.mode==='hide_all_activity'||hasOwnedOutcomes;"
+        in block
+    )
     assert persist_index < return_index
     assert "return true;" not in block
 

tests/test_stable_assistant_turn_anchor_registry.py

@@ -47,6 +47,34 @@ def _function_body(src: str, name: str) -> str:
     raise AssertionError(f"{name} body did not close")
 
 
+def _function_source(src: str, name: str) -> str:
+    start = src.find(f"function {name}")
+    assert start != -1, f"{name} not found"
+    params = src.find("(", start)
+    assert params != -1, f"{name} parameters not found"
+    depth = 0
+    close = -1
+    for idx in range(params, len(src)):
+        if src[idx] == "(":
+            depth += 1
+        elif src[idx] == ")":
+            depth -= 1
+            if depth == 0:
+                close = idx
+                break
+    assert close != -1, f"{name} parameters did not close"
+    brace = src.find("{", close)
+    depth = 0
+    for idx in range(brace, len(src)):
+        if src[idx] == "{":
+            depth += 1
+        elif src[idx] == "}":
+            depth -= 1
+            if depth == 0:
+                return src[start : idx + 1]
+    raise AssertionError(f"{name} body did not close")
+
+
 def _registry_snapshot() -> dict:
     assert NODE, "node is required for assistant_turn_anchors.js registry tests"
     script = f"""
@@ -83,6 +111,7 @@ api.applyAssistantTurnAnchorSourceEvent(registry, {{
 console.log(JSON.stringify({{
   version:api.version,
   registry,
+  scene:api.projectAssistantTurnAnchorActivityScene(registry,{{mode:'compact_worklog'}}),
   isolated,
   results:results.map((item)=>({{applied:item.applied, reason:item.reason}})),
 }}));
@@ -723,6 +752,140 @@ def test_registry_routes_activity_artifacts_side_effects_metadata_and_transport(
     assert anchor["transport_events"][0]["source_event_type"] == "stream_end"
 
 
+def test_live_state_saved_listener_attaches_side_effect_to_anchor():
+    body = _event_listener_body(_read(MESSAGES_JS), "state_saved")
+
+    assert "_applyToAnchor('state_saved',d,e,null,{render:false});" in body
+
+
+def test_invisible_anchor_outcome_applies_without_repainting_live_scene():
+    assert NODE, "node is required for assistant-turn ownership tests"
+    apply_source = _function_source(_read(MESSAGES_JS), "_applyToAnchor")
+    script = f"""
+const activeSid='sid-owned-outcome';
+const streamId='stream-owned-outcome';
+const _anchorRegistry={{anchor:{{}}}};
+const applied=[];
+const _anchorApi={{
+  applyAssistantTurnAnchorSourceEvent(registry,event,context){{
+    applied.push({{registry,event,context}});
+    return {{applied:true}};
+  }},
+}};
+let _anchorShadowWarned=false;
+let _assistantSegmentSeq=3;
+let _currentActivityBurstId=4;
+let renderCount=0;
+function _renderAnchorLiveScene(){{ renderCount+=1; return true; }}
+{apply_source}
+const renderOutcome={{}};
+const result=_applyToAnchor(
+  'state_saved',
+  {{kind:'memory',name:'session-state'}},
+  {{lastEventId:'run-owned-outcome:7'}},
+  renderOutcome,
+  {{render:false}}
+);
+console.log(JSON.stringify({{result,applied,renderCount,renderOutcome}}));
+"""
+    result = subprocess.run(
+        [NODE, "-e", script], text=True, capture_output=True, check=False
+    )
+
+    assert result.returncode == 0, result.stderr
+    data = json.loads(result.stdout)
+    assert data["result"] == {"applied": True}
+    assert data["renderCount"] == 0
+    assert data["renderOutcome"] == {"rendered": False}
+    assert data["applied"][0]["event"]["source_event_type"] == "state_saved"
+    assert data["applied"][0]["event"]["event_id"] == "run-owned-outcome:7"
+    assert data["applied"][0]["context"] == {
+        "session_id": "sid-owned-outcome",
+        "stream_id": "stream-owned-outcome",
+    }
+
+
+def test_side_effect_owner_survives_scene_settlement_and_reload():
+    from api import routes
+
+    data = _registry_snapshot()
+    scene = routes._sanitize_anchor_activity_scene(data["scene"])
+    messages = [
+        {"role": "user", "content": "save this"},
+        {"role": "assistant", "content": "final answer"},
+    ]
+    message_ref = routes._assistant_anchor_scene_message_ref(messages[1])
+    records = {
+        message_ref: {
+            "version": "anchor_activity_scene_record_v1",
+            "message_index": 1,
+            "message_ref": message_ref,
+            "stream_id": "stream-1",
+            "scene": scene,
+        }
+    }
+
+    hydrated = routes._hydrate_anchor_activity_scenes(messages, records)
+    settled_scene = hydrated[1]["_anchor_activity_scene"]
+
+    assert settled_scene["artifacts"] == scene["artifacts"]
+    assert settled_scene["artifacts"][0]["payload"] == {
+        "kind": "workspace_file",
+        "path": "answer.txt",
+    }
+    assert settled_scene["side_effects"] == scene["side_effects"]
+    assert settled_scene["side_effects"][0]["source_event_type"] == "state_saved"
+    assert settled_scene["side_effects"][0]["payload"] == {
+        "kind": "memory",
+        "name": "session-state",
+    }
+
+
+def test_side_effect_only_scene_is_persisted_without_creating_a_worklog():
+    assert NODE, "node is required for assistant-turn ownership tests"
+    messages_js = _read(MESSAGES_JS)
+    has_owned_outcomes = _function_body(messages_js, "_anchorSceneHasOwnedOutcomes")
+    attach_scene = _function_body(
+        messages_js, "_attachProjectedAnchorSceneToLastAssistant"
+    )
+    script = f"""
+const activeSid='sid-owned-outcome';
+const streamId='stream-owned-outcome';
+const _anchorRegistry={{}};
+let persisted=0;
+function _anchorSceneHasOwnedOutcomes(scene){{{has_owned_outcomes}}}
+function _anchorSceneHasWorklogWorthyRows(){{ return false; }}
+function _projectLiveAnchorActivityScene(){{
+  return {{
+    version:'activity_scene_v1',
+    mode:'compact_worklog',
+    activity_rows:[],
+    artifacts:[],
+    side_effects:[{{source_event_type:'state_saved',payload:{{kind:'memory'}}}}],
+  }};
+}}
+function _completeSettledAnchorSceneForTurn(messages,index,scene){{ return scene; }}
+function _persistSettledAnchorScene(){{ persisted+=1; }}
+function _attachProjectedAnchorSceneToLastAssistant(messages){{{attach_scene}}}
+const messages=[{{role:'assistant',content:'final answer'}}];
+const renderedWorklog=_attachProjectedAnchorSceneToLastAssistant(messages);
+console.log(JSON.stringify({{
+  renderedWorklog,
+  persisted,
+  attached:messages[0]._anchor_activity_scene,
+}}));
+"""
+    result = subprocess.run(
+        [NODE, "-e", script], text=True, capture_output=True, check=False
+    )
+
+    assert result.returncode == 0, result.stderr
+    data = json.loads(result.stdout)
+    assert data["renderedWorklog"] is False
+    assert data["persisted"] == 1
+    assert data["attached"]["side_effects"][0]["source_event_type"] == "state_saved"
+
+
 def test_registry_updates_lifecycle_and_settled_final_projection():
     data = _registry_snapshot()
     anchor = data["registry"]["anchor"]
@@ -979,6 +1142,8 @@ def test_activity_scene_is_renderer_neutral_and_empty_safe():
         "final_message_ref": None,
         "terminal_state": None,
         "activity_rows": [],
+        "artifacts": [],
+        "side_effects": [],
     }