Skip to content

emrg: emit anchor-bias-heartbeat with explicit anchored state every round (issue #1078)#1079

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

Merged
argszero merged 1 commit into from
Aug 30, 2026
Merged

emrg: emit anchor-bias-heartbeat with explicit anchored state every round (issue #1078) #1079

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion Agent.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -119,7 +119,7 @@ Community needs voiced in HN agent-UI discussions map directly to EMRG's design:
pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.token; python -m emrg
```

Python: `uv run pytest tests/ -v` (1177) — import check: `uv run python -c "from emrg.client.app import run_client"`
Python: `uv run pytest tests/ -v` (1178) — import check: `uv run python -c "from emrg.client.app import run_client"`
GUI: `cd emrg/gui && npm test` (95: 45 daemon_client + 20 conn-manager + 8 integration + 6 build-config + 7 gui-state + 3 preload-api + 4 boot-contract + 2 theme-guard) — syntax: `node --check main.js preload.js daemon_client.js`
Renderer: `cd emrg/gui/renderer && npm run typecheck && npm test` (479: 5 snapshot-store + 9 utils + 3 ErrorBoundary + 2 App smoke + 11 commands + 4 copywriting + 11 i18n + 11 markdown + 16 transcript + 10 TranscriptView + 15 history + 22 composer + 34 Composer + 6 LinkDialog + 12 sidebar + 17 Sidebar + 9 fileTree + 9 FileTree + 16 resultPanel + 8 ResultPanel + 27 workspaceView + 10 WorkspaceView + 10 dialog + 6 Dialog + 9 ConfirmDialog + 9 RenameDialog + 10 dialogLists + 3 HelpDialog + 9 MemoryDialog + 6 SkillsDialog + 9 openSession + 6 WelcomeDialog + 8 OpenSessionDialog + 7 NewSessionDialog + 7 rewind + 8 RewindDialog + 7 GithubDeviceDialog + 15 daemonBridge + 7 DaemonBridgeProvider + 26 Shell + 15 DialogHost + 20 SettingsPanel + 6 TaskFormDialog + 5 RantDialog + 4 vendorMarkdown) + `npm run build` → `renderer/dist/`
CI: `uv run pytest` (ubuntu + **windows-2025 matrix** — Windows pytest 回归在 PR CI 即失败，v0.2.29 教训 #725) + GUI tests + **actionlint workflow lint** (`rhysd/actionlint@v1.7.12` gate, #444 — workflow 解析错误在 PR CI 即失败，如 `if:` secrets 上下文)
Expand Down
93 changes: 71 additions & 22 deletions emrg/server/daemon.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -2586,24 +2586,15 @@ async def _run_tool_loop(
})
return

# Rant 2026-08-23T13:28:50: refresh the usage anchor from the
# provider's real prompt_tokens + the local estimate of exactly
# what was sent (messages is still the sent set here — assistant
# reply / tool results are appended below). The next round's
# auto-compact projection uses this as its base.
if final_usage:
pt = final_usage.get("prompt_tokens")
if pt:
estimate = self._estimate_tokens(messages)
# Issue #1027: detect a provider/tokenizer silently changing
# under an unchanged base_url/model alias (gateway reroute,
# silent model update) BEFORE the anchor is overwritten —
# the old anchor is the last known same-provider baseline.
self._detect_silent_anchor_drift(session, pt, estimate)
self._usage_anchors[session.session_id] = (
pt, estimate
)
self._record_anchor_drift(session, pt)
# Rant 2026-08-23T13:28:50 + issue #1078: refresh the usage anchor
# from the provider's real prompt_tokens + the local estimate of
# exactly what was sent (messages is still the sent set here —
# assistant reply / tool results are appended below). The next
# round's auto-compact projection uses this as its base. Every
# round emits exactly one anchor-bias-heartbeat with an explicit
# state (anchored=true / anchored=false + reason) so a skipped
# round is a labeled observation, not an absence.
self._refresh_usage_anchor(session, final_usage, messages)

full_content = "".join(content_parts)
# llm.py yields the ACCUMULATED reasoning snapshot on every chunk
Expand DownExpand Up@@ -3091,6 +3082,45 @@ def _record_anchor_drift(self, session, real_pt: int) -> None:
except OSError as exc:
logger.warning("usage-anchor stats append failed: %s", exc)

def _refresh_usage_anchor(self, session, final_usage, messages) -> None:
"""Round-loop usage processing (rant 2026-08-23T13:28:50 + issue
#1078): refresh the usage anchor from the provider's real
prompt_tokens + the local estimate of exactly what was sent. The next
round's auto-compact projection uses this as its base.

Issue #1078 (pm25coder, Dev.to 3dn6k — vinhnguyenthanhdn): the
heartbeat used to live INSIDE ``_detect_silent_anchor_drift``, so a
round that skipped the detector (no usage, no prompt_tokens, no
anchor, invalid estimate) produced zero heartbeat lines — byte-
identical to the detector having stopped. This helper makes every
round emit exactly one ``anchor-bias-heartbeat`` line with an
explicit state: ``anchored=true`` (detector ran, logged inside it) or
``anchored=false`` with a reason. "Log unconditionally, alert
conditionally" now holds one level higher.
"""
if not final_usage:
logger.debug(
"anchor-bias-heartbeat session=%s anchored=false reason=no_usage",
session.session_id,
)
return
pt = final_usage.get("prompt_tokens")
if not pt:
logger.debug(
"anchor-bias-heartbeat session=%s anchored=false "
"reason=no_prompt_tokens",
session.session_id,
)
return
estimate = self._estimate_tokens(messages)
# Issue #1027: detect a provider/tokenizer silently changing under an
# unchanged base_url/model alias (gateway reroute, silent model update)
# BEFORE the anchor is overwritten — the old anchor is the last known
# same-provider baseline.
self._detect_silent_anchor_drift(session, pt, estimate)
self._usage_anchors[session.session_id] = (pt, estimate)
self._record_anchor_drift(session, pt)

def _detect_silent_anchor_drift(self, session, real_pt: int, estimate: int) -> None:
"""Detect a provider/tokenizer silently changing under an unchanged
base_url/model alias (issue #1027, Dev.to 3dicj — heinrichneb: gateway
Expand All@@ -3111,23 +3141,42 @@ def _detect_silent_anchor_drift(self, session, real_pt: int, estimate: int) -> N
Called with the OLD anchor still in place (before the overwrite).
"""
old = self._usage_anchors.get(session.session_id)
if not old or old[1] <= 0 or estimate <= 0:
# Issue #1078: every skip path emits a labeled anchored=false heartbeat
# so a skipped round is a first-class observation, not an absence —
# "log unconditionally, alert conditionally" one level up.
if not old or old[1] <= 0:
logger.debug(
"anchor-bias-heartbeat session=%s anchored=false reason=no_anchor",
session.session_id,
)
return
if estimate <= 0:
logger.debug(
"anchor-bias-heartbeat session=%s anchored=false "
"reason=invalid_estimate",
session.session_id,
)
return
old_real, old_est = old
old_bias = old_real / old_est
new_bias = real_pt / estimate
if old_bias <= 0:
logger.debug(
"anchor-bias-heartbeat session=%s anchored=false reason=invalid_bias",
session.session_id,
)
return
shift = abs(new_bias - old_bias) / old_bias
# Issue #1072 (heinrichneb, Dev.to 3dlo4): make "never fired"
# measurable — log EVERY computed shift unconditionally so the
# detector's heartbeat is visible in emrgd.log (grep
# "anchor-bias-heartbeat"), distinguishing "detector runs, providers
# stable" from "detector dead". The warning below stays reserved for
# actual drift.
# actual drift. Issue #1078: this is the anchored=true case — the
# round produced a real bias measurement.
logger.debug(
"anchor-bias-heartbeat session=%s bias_shift=%.4f threshold=%.2f "
"old_bias=%.3f new_bias=%.3f — %s",
"anchor-bias-heartbeat session=%s anchored=true bias_shift=%.4f "
"threshold=%.2f old_bias=%.3f new_bias=%.3f — %s",
session.session_id, shift, _SILENT_DRIFT_THRESHOLD,
old_bias, new_bias,
"within threshold, no drift"
Expand Down
88 changes: 83 additions & 5 deletions tests/test_daemon.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -807,6 +807,7 @@ def test_silent_provider_drift_emits_event(tmp_path, monkeypatch, caplog):
assert ev["model"] == "gpt-4o-mini" and ev["provider"] == "localhost"
# Heartbeat (issue #1072): unconditional shift log present + DRIFT flag
assert "anchor-bias-heartbeat session=silent-drift" in caplog.text
assert "anchored=true" in caplog.text # issue #1078: anchored state explicit
assert "bias_shift=0.6000" in caplog.text
assert "DRIFT — emitting event" in caplog.text
assert "usage anchor silent drift session=silent-drift" in caplog.text
Expand DownExpand Up@@ -847,23 +848,100 @@ def test_silent_drift_below_threshold_silent(tmp_path, monkeypatch, caplog):
assert ev["provider"] == "localhost"
# Heartbeat present with "within threshold" verdict; no drift event/warning
assert "anchor-bias-heartbeat session=no-drift" in caplog.text
assert "anchored=true" in caplog.text # issue #1078: anchored state explicit
assert "bias_shift=0.0667" in caplog.text
assert "within threshold, no drift" in caplog.text
assert "usage anchor silent drift" not in caplog.text

# Negative states — no observation may be written:
# 1) No anchor at all → no-op (first round, nothing to compare against)
# Negative states — no observation may be written, but issue #1078 makes
# each skipped round a LABELED anchored=false heartbeat (not an absence):
# set_level persists (at_level would restore WARNING on exit, silencing
# the DEBUG heartbeats below).
caplog.set_level("DEBUG", logger="emrg.server.daemon")
# 1) No anchor at all → anchored=false reason=no_anchor (first round)
caplog.clear()
server._detect_silent_anchor_drift(_SidSession("fresh"), 10_000, 9_000)
assert len(stats.read_text(encoding="utf-8").strip().splitlines()) == 1
assert "anchor-bias-heartbeat" not in caplog.text # no old anchor → skip
# 2) Non-positive estimate → no-op (guard: estimate <= 0)
assert ("anchor-bias-heartbeat session=fresh anchored=false "
"reason=no_anchor") in caplog.text
# 2) Non-positive estimate → anchored=false reason=invalid_estimate
caplog.clear()
server._detect_silent_anchor_drift(_SidSession(sid), 240_000, 0)
assert len(stats.read_text(encoding="utf-8").strip().splitlines()) == 1
# 3) Non-positive old bias → no-op (guard: old_bias <= 0)
assert ("anchor-bias-heartbeat session=no-drift anchored=false "
"reason=invalid_estimate") in caplog.text
# 3) Non-positive old bias → anchored=false reason=invalid_bias
caplog.clear()
server._usage_anchors["bad-bias"] = (0, 148_000)
server._detect_silent_anchor_drift(_SidSession("bad-bias"), 240_000, 150_000)
assert len(stats.read_text(encoding="utf-8").strip().splitlines()) == 1
assert ("anchor-bias-heartbeat session=bad-bias anchored=false "
"reason=invalid_bias") in caplog.text


def test_anchor_heartbeat_every_round_labeled(tmp_path, monkeypatch, caplog):
"""Issue #1078 (pm25coder, Dev.to 3dn6k — vinhnguyenthanhdn): the
heartbeat used to live only INSIDE the detector, so a round that skipped
it (no usage / no prompt_tokens) produced zero heartbeat lines —
byte-identical to the detector having stopped. The round-loop helper
``_refresh_usage_anchor`` must emit exactly one labeled heartbeat per
round: ``anchored=false reason=no_usage`` / ``reason=no_prompt_tokens``
on the skip paths, and pass through to the detector (anchored=true) on a
normal round."""
stats = tmp_path / "usage-anchor.jsonl"
monkeypatch.setattr(daemon_mod, "_USAGE_ANCHOR_STATS_PATH", stats)
server = _make_server()
sid = "labeled-rounds"
messages = [
{"role": "system", "content": "sys"},
{"role": "user", "content": "hi"},
]
# Persistent DEBUG for the whole test (at_level would restore WARNING on
# block exit, silencing later heartbeats).
caplog.set_level("DEBUG", logger="emrg.server.daemon")

# 1) No final_usage at all → anchored=false reason=no_usage
server._refresh_usage_anchor(_SidSession(sid), None, messages)
assert ("anchor-bias-heartbeat session=labeled-rounds anchored=false "
"reason=no_usage") in caplog.text
assert "reason=no_prompt_tokens" not in caplog.text
assert sid not in server._usage_anchors # nothing to anchor from

# 2) Empty usage dict → anchored=false reason=no_usage (falsy)
caplog.clear()
server._refresh_usage_anchor(_SidSession(sid), {}, messages)
assert ("anchor-bias-heartbeat session=labeled-rounds anchored=false "
"reason=no_usage") in caplog.text
assert sid not in server._usage_anchors

# 3) Usage present but prompt_tokens missing → reason=no_prompt_tokens
caplog.clear()
server._refresh_usage_anchor(
_SidSession(sid), {"completion_tokens": 5}, messages)
assert ("anchor-bias-heartbeat session=labeled-rounds anchored=false "
"reason=no_prompt_tokens") in caplog.text
assert sid not in server._usage_anchors

# 4) prompt_tokens == 0 (falsy) → reason=no_prompt_tokens
caplog.clear()
server._refresh_usage_anchor(
_SidSession(sid), {"prompt_tokens": 0}, messages)
assert ("anchor-bias-heartbeat session=labeled-rounds anchored=false "
"reason=no_prompt_tokens") in caplog.text
assert sid not in server._usage_anchors

# 5) Normal round → pass-through to detector (anchored=true) + anchor set
caplog.clear()
est = server._estimate_tokens(messages)
server._usage_anchors[sid] = (est, est) # bias 1.0 (self-consistent)
server._refresh_usage_anchor(_SidSession(sid), {"prompt_tokens": est}, messages)
assert "anchored=true" in caplog.text
assert "within threshold, no drift" in caplog.text
anchor = server._usage_anchors[sid]
assert anchor == (est, est) # re-anchored on new real prompt_tokens
lines = stats.read_text(encoding="utf-8").strip().splitlines()
assert len(lines) == 1
assert json.loads(lines[0])["type"] == "anchor_bias_observation"


def test_usage_anchor_cross_provider_window_attributable(tmp_path, monkeypatch):
Expand Down
Loading