Skip to content

fix(pixel): isolate model receipts and preserve managed marker on switches #5579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Lightheartdevs wants to merge 7 commits into
base: public-beta
Choose a base branch
from
Open

fix(pixel): isolate model receipts and preserve managed marker on switches #5579

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
9848bf0
fix(pixel): separate model promotion and route completion receipts
Lightheartdevs Sep 16, 2026
afe7408
fix(pixel): keep managed marker bound to model switches
codex Sep 17, 2026
f608b03
docs(models): scope Granite 3.2 Perplexica warning to Tower1
codex Sep 17, 2026
471bc50
fix(pixel): tolerate validated interrupted transition temp on uninstall
codex Sep 17, 2026
0602e95
Scope Granite H-Tiny OpenCode warning to Tower1
Lightheartdevs Sep 17, 2026
0a59db8
fix(pixel): keep idle local-model streams alive
Lightheartdevs Sep 17, 2026
54999a6
fix(pixel): avoid keepalive comments inside partial SSE lines
Lightheartdevs Sep 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions ods/bin/pixel_access_bridge.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -999,9 +999,22 @@ def remove_model_journal(self):
finally: os.close(directory)

def model_completion(self, request=None):
path = self.state / "model-completed.json"
path = self.state / "model-promotion-completed.json"
legacy = False
if not path.exists():
path = self.state / "model-completed.json"
legacy = True
if not path.exists(): return None
value = private_json(path, 0, 4096)
# Older releases shared this filename with browser model switching.
# A valid receipt from that separate transaction is not a promotion
# completion; malformed state still fails closed.
if legacy and type(value) is dict and set(value) == {"transactionId", "outcome", "configSha256"}:
if (type(value["transactionId"]) is not str or not HEX.fullmatch(value["transactionId"])
or value["outcome"] not in ("commit", "rollback")
or type(value["configSha256"]) is not str or not HEX.fullmatch(value["configSha256"])):
raise AccessError("model-recovery-required")
return None
if (type(value) is not dict
or set(value) != {"kind", "transaction_id", "outcome", "config_sha256"}
or value.get("kind") != "model-completion"
Expand Down Expand Up @@ -1181,7 +1194,7 @@ def model_finish(self, request):
or released_edge.get("streams")):
raise
try:
atomic_json(self.state / "model-completed.json", {
atomic_json(self.state / "model-promotion-completed.json", {
"kind": "model-completion", "transaction_id": pending["transaction_id"],
"outcome": request["outcome"], "config_sha256": config["config_sha256"]})
except OSError:
Expand Down
100 changes: 96 additions & 4 deletions ods/bin/pixel_model_coordinator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@
import hashlib
import json
import os
import stat
import tempfile
import time
from pixel_access_bridge import AccessError, UNIT, atomic_json, private_json, digest, remaining
from pixel_settings.coordinator import _read, _identity, _valid_identity
Expand All @@ -14,7 +16,7 @@ def _sha(value):

def _journal(value):
required = {"kind", "phase", "token", "transactionId", "edge_revision", "edgeHeld", "beforeSha", "afterSha", "target", "boundary", "mode", "beforeIdentity"}
if (type(value) is not dict or set(value) - required - {"outcome"} or not required <= set(value)
if (type(value) is not dict or set(value) - required - {"outcome", "markerBeforeSha"} or not required <= set(value)
or value["kind"] != "model" or value["phase"] not in ("acquiring", "held", "applying", "applied", "restoring", "releasing")
or any(not checksum(value[key]) for key in ("token", "transactionId", "edge_revision", "beforeSha"))
or value["afterSha"] is not None and not checksum(value["afterSha"])
Expand All @@ -24,6 +26,8 @@ def _journal(value):
or type(value["edgeHeld"]) is not bool
or "outcome" in value and value["outcome"] not in ("commit", "rollback")):
raise AccessError("invalid-model-transition")
if "markerBeforeSha" in value and not checksum(value["markerBeforeSha"]):
raise AccessError("invalid-model-transition")
if value["target"] is not None: target(value["target"])
return value

Expand All @@ -36,6 +40,76 @@ def _config(bridge):
return _read(bridge.home / ".openclaw/openclaw.json", bridge.owner.pw_uid)


def _marker_digest(config):
if type(config) is not dict:
raise AccessError("model-marker-invalid")
canonical = json.dumps(config, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(b"ods-pixel-openclaw-v1\0" + canonical).hexdigest()


def _managed_marker(bridge):
path = bridge.home / ".config/ods/pixel-managed.json"
for directory, unsafe_bits in ((path.parent.parent, 0o022), (path.parent, 0o077)):
info = directory.lstat()
if (not stat.S_ISDIR(info.st_mode) or stat.S_ISLNK(info.st_mode)
or info.st_uid != bridge.owner.pw_uid or info.st_mode & unsafe_bits):
raise AccessError("model-marker-unsafe")
marker = private_json(path, bridge.owner.pw_uid, 65536)
if (type(marker) is not dict or marker.get("schema_version") != 2
or marker.get("manager") != "ods" or marker.get("state") != "ready"
or marker.get("initial_active_state") != "absent"
or marker.get("install_dir") != str(bridge.install)
or type(marker.get("configuration_sha256")) is not str
or not checksum(marker["configuration_sha256"])):
raise AccessError("model-marker-invalid")
return path, marker


def _bind_managed_marker(bridge, journal, expected_sha):
before = private_json(bridge.state / "model-before.json", 0, 8 * 1024 * 1024)
prior = _marker_digest(before)
# Pre-upgrade journals did not carry markerBeforeSha. Their root-owned
# model-before snapshot and the still-bound owner marker can prove the
# same prior configuration without silently adopting unrelated drift.
if prior != journal.get("markerBeforeSha", prior):
raise AccessError("model-before-changed")
config, config_sha = _config(bridge)
if config_sha != expected_sha:
raise AccessError("model-config-changed")
path, marker = _managed_marker(bridge)
current = _marker_digest(config)
if marker["configuration_sha256"] == current:
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
try: os.fsync(directory)
finally: os.close(directory)
return # A retry after the marker rename is idempotent.
if marker["configuration_sha256"] != prior:
raise AccessError("model-marker-drifted")
original = path.lstat()
marker["configuration_sha256"] = current
fd, temporary = tempfile.mkstemp(prefix=".pixel-managed.", dir=path.parent)
try:
with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
if os.geteuid() != bridge.owner.pw_uid or os.getegid() != bridge.owner.pw_gid:
os.fchown(handle.fileno(), bridge.owner.pw_uid, bridge.owner.pw_gid)
os.fchmod(handle.fileno(), 0o600)
json.dump(marker, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
observed = path.lstat()
if ((observed.st_dev, observed.st_ino, observed.st_mtime_ns, observed.st_size)
!= (original.st_dev, original.st_ino, original.st_mtime_ns, original.st_size)):
raise AccessError("model-marker-drifted")
os.replace(temporary, path)
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
try: os.fsync(directory)
finally: os.close(directory)
finally:
if os.path.exists(temporary):
os.unlink(temporary)


def _readback(bridge, journal=None):
native = bridge.native()
if native.get("stopped"): raise AccessError("model-runtime-unavailable")
Expand Down Expand Up @@ -142,8 +216,21 @@ def _status(bridge):
raise AccessError("model-runtime-mismatch")
if (_config(bridge)[1] != config_sha or _identity(bridge) != identity or native.get("revision") != observed["revision"]):
raise AccessError("model-inspection-changed")
done_path = bridge.state / "model-completed.json"
done_path = bridge.state / "model-route-completed.json"
legacy = False
if not done_path.exists():
done_path = bridge.state / "model-completed.json"
legacy = True
done = private_json(done_path, 0, 8192) if done_path.exists() else None
# The legacy filename was also used by install-time model promotion.
# Accept its valid receipt as belonging to that other transaction, never
# as evidence that a browser model switch completed.
if legacy and type(done) is dict and set(done) == {"kind", "transaction_id", "outcome", "config_sha256"}:
if (done["kind"] != "model-completion" or not checksum(done["transaction_id"])
or done["outcome"] not in ("applied", "rolled-back")
or not checksum(done["config_sha256"])):
raise AccessError("invalid-model-completion")
done = None
if done and (not checksum(done.get("transactionId")) or done.get("outcome") not in ("commit", "rollback")
or not checksum(done.get("configSha256"))): raise AccessError("invalid-model-completion")
done = done if done and done["configSha256"] == config_sha else None
Expand Down Expand Up @@ -180,9 +267,13 @@ def control(bridge, operation, request=None):
if access["busy"]: raise AccessError("runtime-busy")
if access["configured_mode"] not in ("sandboxed", "full-access"): raise AccessError("model-access-mode-unknown")
config, config_sha = _config(bridge)
_, marker = _managed_marker(bridge)
if marker["configuration_sha256"] != _marker_digest(config):
raise AccessError("model-marker-drifted")
journal = dict(kind="model", phase="acquiring", token=os.urandom(32).hex(), transactionId=request["transactionId"],
edge_revision=access["_edge"]["revision"], edgeHeld=False, beforeSha=config_sha, afterSha=None, target=None,
boundary=bridge.unit_boundary(), mode=access["configured_mode"], beforeIdentity=_identity(bridge))
boundary=bridge.unit_boundary(), mode=access["configured_mode"], beforeIdentity=_identity(bridge),
markerBeforeSha=marker["configuration_sha256"])
atomic_json(bridge.state / "model-before.json", config)
_write(bridge, journal)
_hold(bridge, journal)
Expand Down Expand Up @@ -233,8 +324,9 @@ def control(bridge, operation, request=None):
"config_sha256": expected_sha, "boundary": journal["boundary"]})
_verify(bridge, journal, expected_sha)
if owner["pending"]: worker("model-finish", model_outcome=outcome)
_bind_managed_marker(bridge, journal, expected_sha)
journal.update(phase="releasing", outcome=outcome); _write(bridge, journal)
atomic_json(bridge.state / "model-completed.json", {"transactionId": journal["transactionId"], "outcome": outcome, "configSha256": expected_sha})
atomic_json(bridge.state / "model-route-completed.json", {"transactionId": journal["transactionId"], "outcome": outcome, "configSha256": expected_sha})
bridge.edge("release", journal["token"], journal["edge_revision"])
bridge.native("release", journal["token"])
try:
Expand Down
26 changes: 20 additions & 6 deletions ods/config/model-library.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,6 +356,19 @@
"tokens_per_sec_estimate": 80,
"llm_model_name": "granite-4.0-h-tiny",
"llama_server_image": null,
"app_compatibility": {
"opencode": {
"status": "unsupported_until_revalidated",
"label": "OpenCode revalidation required",
"reason": "In the 2026-09-17 Tower1 exact-head release cycle, OpenCode's build agent issued a read tool call against a fabricated path instead of returning its requested verification phrase on Granite 4.0 H-Tiny. The same installed OpenCode path answered on the baseline and neighboring Granite 4.0 models; Pixel, ODS Talk, LiteLLM, Open WebUI, and Perplexica passed on H-Tiny with exact model-route evidence. Keep this model out of Tower1 all-app release coverage until OpenCode behavior is revalidated.",
"evidence": "ods-public-beta-fleet-green-20260911/release-tower1-pr5579-471bc50-135daf6-r216/model-ui/cycle-002/tower1/model-ui.json",
"recordedAt": "2026-09-17T08:46:03Z",
"productSha": "471bc50df3b3a127aa83ac208c7220e19493d8f6",
"harnessSha": "135daf604399fe757e8364d3721767a22bafe541",
"hostScope": ["tower1"],
"expiresAt": "2026-10-17T00:00:00Z"
}
},
"install_recommendation": false
},
{
Expand Down Expand Up @@ -779,12 +792,13 @@
"perplexica": {
"status": "unsupported_until_revalidated",
"label": "Perplexica revalidation required",
"reason": "Earlier fleet runs found Perplexica nonce failures with this model on Tower2 and M5. A 2026-09-15 Tower3 public-beta run again returned an apologetic Perplexica response without the verification phrase, while ODS Talk, LiteLLM, Open WebUI, and Pixel routed and answered on the selected Granite model. Perplexica answered the same harmless payload after Tower3 recovered its Qwen3.5-27B model. Keep Granite 3.2 2B out of all-app release coverage on these hosts until Perplexica passes a real revalidation.",
"evidence": "fleet-test/runs/2026-07-23T21-41-59Z-release-product-cf6dffb8e8bb-harness-47eedd6c0b40-hosts-six-scope-6cycle-switchboard/model-ui/cycle-005/tower2; fleet-test/runs/2026-07-23T21-41-59Z-release-product-cf6dffb8e8bb-harness-47eedd6c0b40-hosts-six-scope-6cycle-switchboard/model-ui/cycle-005/m5-mbp; ods-public-beta-fleet-green-20260911/release-four-host-2f985137-abb61b2-r46/model-ui/cycle-006/tower3/model-ui.json; Tower3 Perplexica live comparator 2026-09-15T16:59Z",
"recordedAt": "2026-09-15T16:59:00Z",
"productSha": "2f985137b9680151395d59d0498a7fa7e169ff2e",
"harnessSha": "abb61b2586cb347e0a163d4f8feb2acf74e9cee0",
"hostScope": ["tower2", "m5-mbp", "tower3"]
"reason": "Fleet runs on Tower2, M5, Tower3, and Tower1 found Perplexica nonce failures with this model. On the 2026-09-17 Tower1 public-beta candidate, Perplexica returned an apologetic response without the verification phrase while Pixel, OpenCode, LiteLLM, Open WebUI, and ODS Talk answered on the selected Granite model. A Tower3 comparator answered the same harmless payload after restoring Qwen3.5-27B. Keep Granite 3.2 2B out of all-app release coverage on these hosts until Perplexica passes a real revalidation.",
"evidence": "fleet-test/runs/2026-07-23T21-41-59Z-release-product-cf6dffb8e8bb-harness-47eedd6c0b40-hosts-six-scope-6cycle-switchboard/model-ui/cycle-005/tower2; fleet-test/runs/2026-07-23T21-41-59Z-release-product-cf6dffb8e8bb-harness-47eedd6c0b40-hosts-six-scope-6cycle-switchboard/model-ui/cycle-005/m5-mbp; ods-public-beta-fleet-green-20260911/release-four-host-2f985137-abb61b2-r46/model-ui/cycle-006/tower3/model-ui.json; Tower3 Perplexica live comparator 2026-09-15T16:59Z; ods-public-beta-fleet-green-20260911/release-tower1-pr5579-afe7408-ff56b22-r208/model-ui/cycle-006/tower1/model-ui.json",
"recordedAt": "2026-09-17T01:19:42Z",
"productSha": "afe74085e572c167f44563efc68476f1a910e179",
"harnessSha": "ff56b224a69d8b955a23fd24af8407c9ccf6b4e8",
"hostScope": ["tower2", "m5-mbp", "tower3", "tower1"],
"expiresAt": "2026-10-17T00:00:00Z"
},
"hermes_talk": {
"status": "unsupported_until_revalidated",
Expand Down
24 changes: 22 additions & 2 deletions ods/extensions/services/dashboard-api/routers/pixel.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,9 @@
import logging
import os
import re
import time
from pathlib import Path
from typing import AsyncIterator, Literal
from typing import AsyncIterator, Callable, Literal
from urllib.parse import urlparse

import httpx
Expand All @@ -37,6 +38,8 @@
_MODEL = "pixel/default"
_CHAT_STREAM_TIMEOUT_SECONDS = 2040.0
_CLIENT_DISCONNECT_POLL_SECONDS = 0.25
_STREAM_KEEPALIVE_SECONDS = 15.0
_STREAM_KEEPALIVE = b": pixel working\n\n"
_CLIENT_CANCEL_TIMEOUT_SECONDS = 7.0
_MAX_KEY_LENGTH = 4096
_MAX_STATUS_BYTES = 64 * 1024
Expand Down Expand Up @@ -710,6 +713,7 @@ def release(finished):

async def subscribe():
after = -1
last_sent = time.monotonic()
while True:
# Snapshot terminal state before yielding any bytes. Sending a chunk
# can suspend this subscriber while the producer commits its tail.
Expand All @@ -718,10 +722,17 @@ async def subscribe():
for chunk in store.chunks(identity, after):
after = chunk["sequence"]
yield chunk["data"]
last_sent = time.monotonic()
if row is None or row["state"] != "active":
return
if await request.is_disconnected():
return
if time.monotonic() - last_sent >= _STREAM_KEEPALIVE_SECONDS:
# A CPU-backed local model can spend minutes in prompt prefill.
# Keep the subscriber alive without inventing an answer or
# persisting transport-only comments in the result receipt.
yield _STREAM_KEEPALIVE
last_sent = time.monotonic()
# Subscriber disposal never cancels the independent bounded producer.
await asyncio.sleep(_CLIENT_DISCONNECT_POLL_SECONDS)

Expand Down Expand Up @@ -850,10 +861,12 @@ def _edge_chat_body(body, messages):
async def _iter_upstream_chunks(
upstream: httpx.Response,
request: Request,
can_emit_keepalive: Callable[[], bool],
) -> AsyncIterator[bytes]:
"""Yield upstream bytes while promptly observing a silent client exit."""
iterator = upstream.aiter_bytes().__aiter__()
pending: asyncio.Task[bytes] | None = None
last_sent = time.monotonic()
try:
while True:
pending = asyncio.create_task(anext(iterator))
Expand All @@ -866,12 +879,19 @@ async def _iter_upstream_chunks(
break
if await request.is_disconnected():
raise _ClientDisconnected
# A comment is safe only between complete SSE lines. The
# caller may be holding an upstream fragment without a newline;
# injecting a comment there would corrupt that data line.
if can_emit_keepalive() and time.monotonic() - last_sent >= _STREAM_KEEPALIVE_SECONDS:
yield _STREAM_KEEPALIVE
last_sent = time.monotonic()
try:
chunk = pending.result()
except StopAsyncIteration:
return
pending = None
yield chunk
last_sent = time.monotonic()
finally:
if pending is not None and not pending.done():
pending.cancel()
Expand Down Expand Up @@ -977,7 +997,7 @@ async def stream() -> AsyncIterator[bytes]:
try:
async with async_timeout(_CHAT_STREAM_TIMEOUT_SECONDS):
buffered = bytearray()
async for chunk in _iter_upstream_chunks(upstream, request):
async for chunk in _iter_upstream_chunks(upstream, request, lambda: not buffered):
buffered.extend(chunk)
while True:
newline = buffered.find(b"\n")
Expand Down
8 changes: 7 additions & 1 deletion ods/extensions/services/dashboard-api/tests/test_performance_oracle.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -588,7 +588,7 @@ def test_real_catalog_gemma_perplexica_block_is_global():
assert tower2["perplexica"]["status"] == "unsupported_until_revalidated"


def test_real_catalog_granite32_perplexica_block_includes_tower3_after_live_failure():
def test_real_catalog_granite32_perplexica_block_includes_tower1_and_tower3_after_live_failure():
by_id = {model["id"]: model for model in _official_model_catalog()}
model = by_id["granite3.2-2b-instruct-q4"]

Expand All @@ -612,13 +612,19 @@ def test_real_catalog_granite32_perplexica_block_includes_tower3_after_live_fail
model,
runtime_context={"host": "tower3", "hosts": ["tower3"]},
)
tower1 = model_app_compatibility(
model,
runtime_context={"host": "tower1", "hosts": ["tower1"]},
)

assert windows_laptop["perplexica"]["status"] == "unknown"
assert strix_halo["perplexica"]["status"] == "unknown"
assert tower2["perplexica"]["status"] == "unsupported_until_revalidated"
assert m5_mbp["perplexica"]["status"] == "unsupported_until_revalidated"
assert tower3["perplexica"]["status"] == "unsupported_until_revalidated"
assert tower1["perplexica"]["status"] == "unsupported_until_revalidated"
assert "Tower3" in tower3["perplexica"]["reason"]
assert "Tower1" in tower1["perplexica"]["reason"]


def test_real_catalog_smollm3_perplexica_block_is_global():
Expand Down
Loading
Loading