GodopuandClaude Opus 5
fb3498e23b
The four delegation scripts become a real package. Their old paths stay as
shims because those paths are a wire contract: agent instructions, the bash
wrapper, SKILL.md, BOOTSTRAP.md, deploy/install.sh and three test modules all
call them literally.
The shim binds the lib_py.delegation *package* to the tree beside it, once per
process, and never touches sys.path. Two cheaper designs were rejected with
evidence (the context matrix now lives in the test suite):
* inserting the sibling skills dir at sys.path[0] makes the first sandbox
copy imported under pytest shadow lib_py.agents/layout/state for the whole
session, and the identity assertion cannot see it -- `is` proves same
object, not same tree.
* loading only the named module by file path leaves a module that imports
siblings pulling them from whichever tree is on the path, so a frozen
publish_event ran against live dependencies and lost half of B-13.
Binding the package fixes both: one tree for all delegation modules, the parent
lib_py untouched, and a sys.path append only when no lib_py exists anywhere
(true standalone). Bind-once rather than nearest-wins, because rebinding would
split existing references and break monkeypatching.
Also moves the env-file load off import (_load_dotenv -> ensure_env_loaded,
called by each process entry point including reconcile.sh's embedded
subscriber), so importing the library no longer mutates os.environ.
Verified: PV-1 identity+provenance, PV-4's ten import scenarios plus K1/K3/K6
in fresh interpreters, PV-5 shim byte-identity, an AST check that the package
imports no sibling lib_py module, and --help output diffed against the
pre-move capture.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
127 lines
5.0 KiB
Python
127 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Shim context harness (PV-4, issue #5 phase 1).
|
|
|
|
Runs ONE import scenario in a fresh interpreter and prints a JSON report of
|
|
which tree every relevant module resolved to. Kept out of the test module on
|
|
purpose: each scenario needs a virgin ``sys.modules``, so it must be a
|
|
subprocess.
|
|
|
|
Usage: shim_contexts.py <scenario> <real> <sandbox> <sandbox2> <freeze>
|
|
|
|
Tree arguments are skills-directory roots, each containing ``lib_py/`` and
|
|
``multi-agent-mux-delegate-job/scripts/``. The scenarios mirror the four real
|
|
execution contexts: pytest in-process (A/A_pe/B/B2/D/E), standalone with no
|
|
ambient tree (C1/C1_pe), and a frozen copy invoked while PYTHONPATH still points
|
|
at the live tree (C2/C2_pe) -- the shape every delegated agent job actually runs.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
SCENARIO = sys.argv[1]
|
|
TREES = {
|
|
"real": os.path.abspath(sys.argv[2]),
|
|
"sandbox": os.path.abspath(sys.argv[3]),
|
|
"sandbox2": os.path.abspath(sys.argv[4]),
|
|
"freeze": os.path.abspath(sys.argv[5]),
|
|
}
|
|
|
|
|
|
def scripts(tree):
|
|
return os.path.join(TREES[tree], "multi-agent-mux-delegate-job", "scripts")
|
|
|
|
|
|
def which_tree(path):
|
|
if not path:
|
|
return None
|
|
path = os.path.abspath(path)
|
|
for name, root in TREES.items():
|
|
if path == root or path.startswith(root + os.sep):
|
|
return name
|
|
return "OUTSIDE:" + path
|
|
|
|
|
|
def report():
|
|
out = {"scenario": SCENARIO}
|
|
mods = sys.modules
|
|
|
|
def note(key, mod, attr="__file__"):
|
|
m = mods.get(mod)
|
|
if m is None:
|
|
out[key] = None
|
|
return
|
|
if attr == "__path__":
|
|
out[key] = which_tree(list(getattr(m, "__path__", [None]))[0])
|
|
else:
|
|
out[key] = which_tree(getattr(m, attr, None))
|
|
|
|
note("alias_mqtt_common", "mqtt_common")
|
|
note("alias_publish_event", "publish_event")
|
|
note("pkg_delegation", "lib_py.delegation", "__path__")
|
|
note("pkg_mqtt_common", "lib_py.delegation.mqtt_common")
|
|
note("pkg_publish_event", "lib_py.delegation.publish_event")
|
|
note("lib_py", "lib_py", "__path__")
|
|
note("lib_py_agents", "lib_py.agents", "__path__")
|
|
|
|
# identity: the alias and the package attribute must be one object
|
|
ident = {}
|
|
for name in ("mqtt_common", "publish_event"):
|
|
alias = mods.get(name)
|
|
pkg_mod = mods.get("lib_py.delegation." + name)
|
|
ident[name] = None if alias is None else (alias is pkg_mod)
|
|
out["identity"] = ident
|
|
|
|
# PV-3 shape: sys.path entries that would contribute a lib_py package
|
|
out["lib_py_path_entries"] = [
|
|
which_tree(e) for e in sys.path
|
|
if e and os.path.isdir(os.path.join(e, "lib_py"))
|
|
]
|
|
return out
|
|
|
|
|
|
if SCENARIO in ("A", "A_pe", "B", "B2", "D", "E"):
|
|
sys.path.insert(0, TREES["real"]) # models conftest.py inserting the real tree
|
|
|
|
if SCENARIO == "A":
|
|
sys.path.insert(0, scripts("sandbox"))
|
|
import mqtt_common, publish_event # noqa: F401
|
|
elif SCENARIO == "A_pe":
|
|
sys.path.insert(0, scripts("sandbox"))
|
|
import publish_event # noqa: F401
|
|
elif SCENARIO == "B":
|
|
import lib_py.agents # noqa: F401
|
|
sys.path.insert(0, scripts("sandbox"))
|
|
import mqtt_common, publish_event # noqa: F401
|
|
elif SCENARIO == "B2":
|
|
import lib_py.delegation.mqtt_common # noqa: F401
|
|
sys.path.insert(0, scripts("sandbox"))
|
|
import mqtt_common, publish_event # noqa: F401
|
|
elif SCENARIO == "D":
|
|
sys.path.insert(0, scripts("sandbox"))
|
|
import mqtt_common # noqa: F401
|
|
sys.path.insert(0, scripts("sandbox2"))
|
|
import publish_event # noqa: F401
|
|
elif SCENARIO == "E":
|
|
sys.path.insert(0, scripts("sandbox"))
|
|
import mqtt_common # noqa: F401
|
|
from lib_py.delegation import publish_event # noqa: F401
|
|
elif SCENARIO == "C1":
|
|
sys.path.insert(0, scripts("freeze"))
|
|
import mqtt_common, publish_event # noqa: F401
|
|
elif SCENARIO == "C1_pe":
|
|
sys.path.insert(0, scripts("freeze"))
|
|
import publish_event # noqa: F401
|
|
elif SCENARIO == "C2":
|
|
sys.path.insert(0, scripts("freeze")) # PYTHONPATH=real set by caller
|
|
import mqtt_common, publish_event # noqa: F401
|
|
elif SCENARIO == "C2_pe":
|
|
sys.path.insert(0, scripts("freeze")) # PYTHONPATH=real set by caller
|
|
import publish_event # noqa: F401
|
|
elif SCENARIO == "K3":
|
|
sys.path.insert(0, scripts("real"))
|
|
import mqtt_common, publish_event # noqa: F401
|
|
else:
|
|
raise SystemExit("unknown scenario: " + SCENARIO)
|
|
|
|
print(json.dumps(report()))
|