from __future__ import annotations
import json
import unittest
from peer import Peer, output
class EnvironmentDiscoverabilityTests(unittest.IsolatedAsyncioTestCase):
async def test_inspection_uses_host_manifest_not_worker_cwd(self):
"""The normal SDK must expose the host response, not infer identity locally."""
host_manifest = {
"source_root": {"path": "/truthful/workspace", "exists": True, "writable": False},
"draft_root": {"path": "/truthful/workspace/.klbr/drafts", "target": {"id": "target-remote", "workspace": "/truthful/workspace"}, "exists": False},
"target": {"id": "target-remote", "workspace": "/truthful/workspace"},
"selected_behavior_revision": "release-selected",
"selected_behavior_epoch": 7,
"applied_behavior_revision": "release-applied",
"purpose": "interactive",
"cell_budget": {"limit_ms": 1700, "remaining_ms": 1337},
}
async def service(message):
self.assertEqual(message["request"], {"method": "runtime.inspect"})
return {"kind": "ok", "value": host_manifest}
async with Peer() as peer:
result = await peer.execute(
"from klbr import runtime\nimport json\nprint(json.dumps(await runtime.inspect(), sort_keys=True))",
service,
)
self.assertEqual(result["status"], "ok")
inspected = json.loads(output(result))
self.assertEqual(inspected, host_manifest)
self.assertNotEqual(inspected["source_root"]["path"], str(peer.directory))
self.assertEqual(inspected["draft_root"]["path"], "/truthful/workspace/.klbr/drafts")
self.assertEqual(inspected["selected_behavior_revision"], "release-selected")
self.assertEqual(inspected["applied_behavior_revision"], "release-applied")
self.assertNotIn("api_key", inspected)
secret = "credential-value-must-not-appear"
self.assertNotIn(secret, json.dumps(inspected))
async def test_inspection_does_not_turn_unavailable_fields_into_guesses(self):
async def service(message):
self.assertEqual(message["request"]["method"], "runtime.inspect")
return {
"kind": "ok",
"value": {
"draft_root": {"path": "/truthful/workspace/.klbr/drafts", "exists": False},
"unavailable": {"cumulative_model_budget": "not supplied"},
},
}
async with Peer() as peer:
result = await peer.execute(
"from klbr import runtime\nimport json\nprint(json.dumps(await runtime.inspect()))",
service,
)
inspected = json.loads(output(result))
self.assertEqual(inspected["draft_root"]["exists"], False)
self.assertNotIn("draft_root", inspected["unavailable"])
self.assertNotIn("cwd", inspected)
self.assertNotIn("ps", inspected)
if __name__ == "__main__":
unittest.main()