tmpl/multi-agent-mux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(config): resolve dotenv via workspace marker and prefer .mam.env over legacy .env

Browse Source
This commit is contained in:
Godopu
2026-08-04 21:42:45 +09:00
parent 11583eb173
commit 2458995e75
1 changed files with 69 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.agents/skills/multi-agent-mux-delegate-job/scripts/mqtt_common.py
+69 -25
View File
@@ -33,37 +33,81 @@ import paho.mqtt.client as mqtt
logger = logging.getLogger("delegate_job.mqtt_common")
def _load_dotenv(workspace_dir: str = None) -> None:
"""Load .env file from workspace if it exists and env var not already set.
_warned_deprecated_env = False
_warned_coexistence_env = False
This ensures Python scripts get the same env vars as the shell wrapper
scripts that source .env. Only sets vars that are not already in os.environ
(i.e. OS env takes precedence over .env file).
def _load_dotenv(workspace_dir: Optional[str] = None) -> None:
"""Load .mam.env (or .env fallback) from workspace if it exists.
Only sets vars that are not already in os.environ
(i.e. OS env takes precedence over env files).
"""
import os
global _warned_deprecated_env, _warned_coexistence_env
# 1. Check explicit MAM_ENV_FILE override
explicit_file = os.environ.get("MAM_ENV_FILE")
if explicit_file:
if os.path.isfile(explicit_file):
_parse_env_file(explicit_file)
return
# 2. Resolve workspace directory with boundary marker check (.agents or .git)
if workspace_dir is None:
# Walk up from this script to find workspace root
d = os.path.dirname(os.path.abspath(__file__))
for _ in range(5):
if os.path.isfile(os.path.join(d, ".env")):
curr = os.path.dirname(os.path.abspath(__file__))
resolved_root = None
while curr and curr != os.path.dirname(curr):
if os.path.isdir(os.path.join(curr, ".agents")) or os.path.exists(os.path.join(curr, ".git")):
resolved_root = curr
break
d = os.path.dirname(d)
curr = os.path.dirname(curr)
if not resolved_root:
return
d = resolved_root
else:
d = workspace_dir
env_path = os.path.join(d, ".env")
if not os.path.isfile(env_path):
return
with open(env_path, "r") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, val = line.partition("=")
key = key.strip()
val = val.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = val
mam_env_path = os.path.join(d, ".mam.env")
legacy_env_path = os.path.join(d, ".env")
has_mam = os.path.isfile(mam_env_path)
has_legacy = os.path.isfile(legacy_env_path)
if has_mam and has_legacy:
if not _warned_coexistence_env:
logger.warning(
"Both '%s' and '%s' exist. Loading '%s'. "
"Consider removing or migrating '%s'.",
mam_env_path, legacy_env_path, mam_env_path, legacy_env_path
)
_warned_coexistence_env = True
_parse_env_file(mam_env_path)
elif has_mam:
_parse_env_file(mam_env_path)
elif has_legacy:
if not _warned_deprecated_env:
logger.warning(
"Loading deprecated config file '%s'. "
"Please migrate to '.mam.env'.",
legacy_env_path
)
_warned_deprecated_env = True
_parse_env_file(legacy_env_path)
def _parse_env_file(path: str) -> None:
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, val = line.partition("=")
key = key.strip()
val = val.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = val
except Exception as e:
logger.warning("Failed to parse env file %s: %s", path, e)
_load_dotenv()