Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
37 changes: 33 additions & 4 deletions agent/conversation_loop.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import os
import random
import re
import socket
import ssl
import sys
import time
Expand Down Expand Up @@ -1258,16 +1259,17 @@ def host_info_value(label: str) -> str:
each message and destroying the prefix cache for the whole session,
which is far worse than the staleness this function guards against.

Anchor on the ``User home directory:`` line that immediately precedes
the working-directory line in that block, and take the FIRST such
occurrence, so only Hermes' own emitted block can satisfy the read.
Anchor on the ``User home directory:`` line in the middle of that
block, and take the FIRST such occurrence, so only Hermes' own emitted
block can satisfy the read. Runtime host fields precede the anchor;
the working-directory field follows it.
"""
prefix = f"{label}:"
lines = prompt.splitlines()
for idx, line in enumerate(lines):
if not line.startswith("User home directory:"):
continue
for candidate in lines[idx + 1: idx + 4]:
for candidate in lines[max(0, idx - 3): idx + 4]:
if candidate.startswith(prefix):
return candidate[len(prefix):].strip()
return ""
Expand All @@ -1292,6 +1294,33 @@ def host_info_value(label: str) -> str:
if stored_cwd != str(resolve_agent_cwd()):
return False

# Session databases can be synchronized between machines. A stored local
# prompt must never carry the other machine's runtime identity into this
# process merely because its profile name and cwd happen to match. Legacy
# local prompts have no hostname at all, so reject them once to upgrade the
# persisted snapshot. Remote/sandbox prompts intentionally omit the whole
# host-info block and remain reusable.
#
# Hostname is a useful human-readable signal but not a globally unique
# gateway identity: two cloned VMs or machines can share a hostname, and
# two gateway instances can run on the same host. This is host anchoring,
# not an authentication boundary; track a follow-up for a stable opaque
# per-install/runtime identity if stronger guarantees are needed.
stored_host = host_info_value("Host")
stored_hostname = host_info_value("Machine hostname")
if stored_host:
if not stored_hostname:
# Legacy local prompt without hostname: reject once to upgrade.
return False
try:
current_hostname = socket.gethostname().strip()
except OSError:
current_hostname = ""
# Fail closed: a local stored prompt must not silently be treated as
# verified when the live hostname cannot be read.
if not current_hostname or stored_hostname.casefold() != current_hostname.casefold():
return False

# Detect runtime-surface drift: the stored prompt records which platform it
# was built for (e.g. "desktop" vs "cli"). Reusing a desktop-built prompt on
# a terminal session (or vice versa) would inject the wrong runtime hints.
Expand Down
11 changes: 9 additions & 2 deletions agent/prompt_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import logging
import os
import queue
import socket
import sys
import threading
import contextvars
Expand Down Expand Up @@ -1406,8 +1407,8 @@ def build_environment_hints() -> str:
"""Return environment-specific guidance for the system prompt.

Always emits a factual block describing the execution environment:
- For **local** terminal backends: the host OS, user home, current
working directory (plus a Windows-only note about hostname != user
- For **local** terminal backends: the host OS, live machine hostname,
user home, current working directory (plus a Windows-only note about hostname != user
and a Windows-only note that `terminal` shells out to bash, not
PowerShell).
- For **remote / sandbox** terminal backends (docker, singularity,
Expand Down Expand Up @@ -1439,6 +1440,12 @@ def build_environment_hints() -> str:
else:
host_lines.append(f"Host: {platform.system()} ({platform.release()})")

try:
hostname = socket.gethostname().strip()
except OSError:
hostname = ""
if hostname:
host_lines.append(f"Machine hostname: {hostname}")
host_lines.append(f"User home directory: {os.path.expanduser('~')}")
try:
host_lines.append(f"Current working directory: {resolve_agent_cwd()}")
Expand Down
39 changes: 39 additions & 0 deletions tests/agent/test_prompt_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -763,6 +763,45 @@ def test_build_environment_hints_suppresses_host_on_docker_backend(self, monkeyp
assert "Terminal backend: docker" in result
assert "inside" in result.lower()

def test_build_environment_hints_identifies_live_machine(self, monkeypatch):
"""The prompt must identify the process host, not synchronized state."""
import agent.prompt_builder as _pb

monkeypatch.setenv("TERMINAL_ENV", "local")
monkeypatch.setattr("socket.gethostname", lambda: "atlas")

result = _pb.build_environment_hints()

assert "Machine hostname: atlas" in result
assert result.index("Machine hostname: atlas") < result.index("User home directory:")

def test_build_environment_hints_omits_hostname_when_gethostname_raises(self, monkeypatch):
"""Regression: if socket.gethostname() raises OSError, the prompt must still build."""
import agent.prompt_builder as _pb

def _raise_oserror():
raise OSError("gethostname failed")

monkeypatch.setenv("TERMINAL_ENV", "local")
monkeypatch.setattr("socket.gethostname", _raise_oserror)

result = _pb.build_environment_hints()

assert "Machine hostname:" not in result
assert "User home directory:" in result

def test_build_environment_hints_omits_hostname_when_gethostname_returns_empty(self, monkeypatch):
"""Regression: if socket.gethostname() returns empty string, omit the hostname line."""
import agent.prompt_builder as _pb

monkeypatch.setenv("TERMINAL_ENV", "local")
monkeypatch.setattr("socket.gethostname", lambda: "")

result = _pb.build_environment_hints()

assert "Machine hostname:" not in result
assert "User home directory:" in result

def test_build_environment_hints_uses_terminal_cwd_over_launch_dir(self, monkeypatch, tmp_path):
"""THE BUG: gateway/cron set TERMINAL_CWD but the prompt emitted os.getcwd()
(the daemon launch dir). Regression for #24882/#24969/#27383/#29265."""
Expand Down
101 changes: 100 additions & 1 deletion tests/run_agent/test_compression_persistence.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
"""

import os
import socket
import tempfile
from pathlib import Path
from unittest.mock import patch
Expand Down Expand Up @@ -390,7 +391,7 @@ class _Agent:
return agent

@staticmethod
def _host_block(cwd: str) -> str:
def _host_block(cwd: str, hostname: str | None = None) -> str:
"""A stored prompt fragment shaped like the real host-info block.

``build_environment_hints`` always emits ``User home directory:``
Expand All @@ -401,10 +402,108 @@ def _host_block(cwd: str) -> str:
"""
return (
"Host: Linux (6.16.0)\n"
f"Machine hostname: {hostname or socket.gethostname()}\n"
"User home directory: /home/tester\n"
f"Current working directory: {cwd}\n"
)

def test_stored_prompt_stale_when_machine_hostname_differs(self):
"""A session DB copied from another host must rebuild its prompt."""
from unittest.mock import patch
from agent.conversation_loop import _stored_prompt_matches_runtime

agent = self._make_agent()
stored_prompt = self._host_block("/project/current", hostname="personal-pc")

with (
patch("agent.conversation_loop.resolve_agent_cwd", return_value="/project/current"),
patch("socket.gethostname", return_value="atlas"),
):
assert _stored_prompt_matches_runtime(agent, stored_prompt) is False

def test_stored_prompt_fresh_when_machine_hostname_matches(self):
from unittest.mock import patch
from agent.conversation_loop import _stored_prompt_matches_runtime

agent = self._make_agent()
stored_prompt = (
self._host_block("/project/current", hostname="atlas")
+ "\n# AGENTS.md\nMachine hostname: personal-pc\n"
)

with (
patch("agent.conversation_loop.resolve_agent_cwd", return_value="/project/current"),
patch("socket.gethostname", return_value="atlas"),
):
assert _stored_prompt_matches_runtime(agent, stored_prompt) is True

def test_legacy_local_prompt_without_machine_hostname_is_rebuilt(self):
"""One rebuild upgrades pre-runtime-identity prompt snapshots."""
from unittest.mock import patch
from agent.conversation_loop import _stored_prompt_matches_runtime

agent = self._make_agent()
stored_prompt = (
"Host: Linux (6.16.0)\n"
"User home directory: /home/tester\n"
"Current working directory: /project/current\n"
)

with (
patch("agent.conversation_loop.resolve_agent_cwd", return_value="/project/current"),
patch("socket.gethostname", return_value="atlas"),
):
assert _stored_prompt_matches_runtime(agent, stored_prompt) is False

def test_stored_prompt_rejected_when_gethostname_raises_oserror(self):
"""Fail closed: if socket.gethostname() raises OSError, reject the stored prompt."""
from unittest.mock import patch
from agent.conversation_loop import _stored_prompt_matches_runtime

agent = self._make_agent()
stored_prompt = self._host_block("/project/current", hostname="atlas")

with (
patch("agent.conversation_loop.resolve_agent_cwd", return_value="/project/current"),
patch("socket.gethostname", side_effect=OSError("gethostname failed")),
):
assert _stored_prompt_matches_runtime(agent, stored_prompt) is False

def test_stored_prompt_rejected_when_gethostname_returns_empty(self):
"""Fail closed: if socket.gethostname() returns empty string, reject the stored prompt."""
from unittest.mock import patch
from agent.conversation_loop import _stored_prompt_matches_runtime

agent = self._make_agent()
stored_prompt = self._host_block("/project/current", hostname="atlas")

with (
patch("agent.conversation_loop.resolve_agent_cwd", return_value="/project/current"),
patch("socket.gethostname", return_value=""),
):
assert _stored_prompt_matches_runtime(agent, stored_prompt) is False

def test_same_hostname_different_profile_still_valid(self):
"""Hostname is host anchoring, not authentication: same host, different profile is valid."""
from unittest.mock import patch
from agent.conversation_loop import _stored_prompt_matches_runtime

agent = self._make_agent()
# Same hostname but different user home (different profile)
stored_prompt = (
"Host: Linux (6.16.0)\n"
"Machine hostname: atlas\n"
"User home directory: /home/other-profile\n"
"Current working directory: /project/current\n"
)

with (
patch("agent.conversation_loop.resolve_agent_cwd", return_value="/project/current"),
patch("socket.gethostname", return_value="atlas"),
):
# Same hostname should be considered valid even with different profile
assert _stored_prompt_matches_runtime(agent, stored_prompt) is True

def test_stored_prompt_stale_when_cwd_differs(self):
"""Different cwd should force a prompt rebuild."""
from unittest.mock import patch
Expand Down
Loading