Refactor qa/sim runtime to config-first task flow

Author: zh1peng

psyflow/StimUnit.py

@@ -3,7 +3,7 @@
 from typing import Callable, Optional, List, Dict, Any, Sequence, TypeAlias, Union
 import importlib
 import random
-from .qa.context import get_context
+from .sim.context import get_context
 from .io.events import TriggerEvent
 from .sim.adapter import ResponderAdapter, ResponderActionError
 from .sim.contracts import Feedback, Observation
@@ -83,7 +83,7 @@ def _qa_scale_duration(self, nominal_s: float) -> tuple[float, int, bool]:
         """Return (used_seconds, n_frames, scaled_flag) for QA mode.
 
         This keeps default behavior unchanged. In QA mode, scaling is opt-in
-        via QAContext.config.enable_scaling.
+        via runtime context config.enable_scaling.
         """
         used = float(nominal_s)
         n_frames = max(1, int(round(used / self.frame_time)))

psyflow/__init__.py

@@ -38,6 +38,17 @@
     "initialize_triggers": ("psyflow.io", "initialize_triggers"),
     "initialize_exp": ("psyflow.utils", "initialize_exp"),
     "list_supported_voices": ("psyflow.utils", "list_supported_voices"),
+    # Task runtime option parsing helpers
+    "TaskRunOptions": ("psyflow.task_options", "TaskRunOptions"),
+    "build_task_arg_parser": ("psyflow.task_options", "build_task_arg_parser"),
+    "parse_task_run_options": ("psyflow.task_options", "parse_task_run_options"),
+    "resolve_config_path": ("psyflow.task_options", "resolve_config_path"),
+    "resolve_mode": ("psyflow.task_options", "resolve_mode"),
+    # Sim/QA runtime context helpers
+    "RuntimeContext": ("psyflow.sim", "RuntimeContext"),
+    "context_from_config": ("psyflow.sim", "context_from_config"),
+    "runtime_context": ("psyflow.sim", "runtime_context"),
+    "set_trial_context": ("psyflow.sim", "set_trial_context"),
 }
 
 __all__ = ["__version__", *_LAZY_ATTRS.keys()]
@@ -51,6 +62,19 @@
     from .TaskSettings import TaskSettings as TaskSettings
     from .cli import main as cli_main
     from .io import initialize_triggers as initialize_triggers
+    from .task_options import (
+        TaskRunOptions as TaskRunOptions,
+        build_task_arg_parser as build_task_arg_parser,
+        parse_task_run_options as parse_task_run_options,
+        resolve_config_path as resolve_config_path,
+        resolve_mode as resolve_mode,
+    )
+    from .sim import (
+        RuntimeContext as RuntimeContext,
+        context_from_config as context_from_config,
+        runtime_context as runtime_context,
+        set_trial_context as set_trial_context,
+    )
     from .utils import (
         count_down as count_down,
         initialize_exp as initialize_exp,

psyflow/io/runtime.py

@@ -65,7 +65,7 @@ def _next_id(self) -> int:
     def _log(self, rec: dict[str, Any]) -> None:
         # 1) QA event sink (if active)
         try:
-            from psyflow.qa.context import log_event
+            from psyflow.sim.context import log_event
 
             log_event(rec)
         except Exception:
@@ -90,7 +90,7 @@ def _effective_strict(self) -> bool:
         if self.strict:
             return True
         try:
-            from psyflow.qa.context import get_context
+            from psyflow.sim.context import get_context
 
             ctx = get_context()
             return bool(ctx is not None and getattr(getattr(ctx, "config", None), "strict", False))

psyflow/qa/__init__.py

@@ -4,28 +4,17 @@
 used in lightweight static checks and CI.
 """
 
-from .context import QAConfig, QAContext, context_from_env, get_context, log_event, log_sim_event, qa_context
-from .responder import NullResponder, ScriptedResponder
-from .static import contract_lint, load_yaml, static_qa
-from .trace import validate_events, validate_trace_csv
 from .artifacts import qa_artifact_paths
 from .report import FailureCode, QAReport
+from .static import contract_lint, load_yaml, static_qa
+from .trace import validate_events, validate_trace_csv
 
 __all__ = [
     "FailureCode",
-    "QAConfig",
-    "QAContext",
     "QAReport",
-    "NullResponder",
-    "ScriptedResponder",
-    "context_from_env",
     "contract_lint",
-    "get_context",
     "load_yaml",
-    "log_event",
-    "log_sim_event",
     "qa_artifact_paths",
-    "qa_context",
     "static_qa",
     "validate_events",
     "validate_trace_csv",

psyflow/qa/responder.py

@@ -17,6 +17,15 @@
     ScriptedResponder,
     SessionInfo,
 )
+from .context import (
+    RuntimeConfig,
+    RuntimeContext,
+    context_from_config,
+    get_context,
+    log_event,
+    log_sim_event,
+    runtime_context,
+)
 from .context_helpers import set_trial_context
 from .loader import load_responder
 from .logging import iter_sim_events, make_sim_jsonl_logger
@@ -31,12 +40,19 @@
     "ResponderActionError",
     "ResponderAdapter",
     "ResponderProtocol",
+    "RuntimeConfig",
+    "RuntimeContext",
     "ScriptedResponder",
     "SessionInfo",
+    "context_from_config",
+    "get_context",
+    "log_event",
+    "log_sim_event",
     "load_responder",
     "iter_sim_events",
     "make_rng",
     "make_sim_jsonl_logger",
     "make_trial_seed",
+    "runtime_context",
     "set_trial_context",
 ]

psyflow/sim/__init__.py

@@ -3,16 +3,15 @@
 import contextlib
 import contextvars
 import json
-import os
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Callable, Optional
 
-from psyflow.sim.contracts import SessionInfo
-from psyflow.sim.loader import load_responder
-from psyflow.sim.logging import make_sim_jsonl_logger
-from psyflow.sim.rng import make_rng
+from .contracts import SessionInfo
+from .loader import load_responder
+from .logging import make_sim_jsonl_logger
+from .rng import make_rng
 
 
 def _cfg_get(mapping: dict[str, Any] | None, path: tuple[str, ...], default: Any = None) -> Any:
@@ -40,10 +39,13 @@ def _as_bool(value: Any, default: bool = False) -> bool:
     return bool(default)
 
 
-@dataclass(frozen=True)
-class QAConfig:
-    """QA/sim knobs used by the responder injection layer."""
+def _normalize_mode(value: Any, default: str = "human") -> str:
+    mode = str(value or "").strip().lower() or str(default or "human").strip().lower() or "human"
+    return mode if mode in ("human", "qa", "sim") else "human"
+
 
+@dataclass(frozen=True)
+class RuntimeConfig:
     enable_scaling: bool = False
     timing_scale: float = 1.0
     min_frames: int = 2
@@ -55,11 +57,11 @@ class QAConfig:
 
 
 @dataclass
-class QAContext:
+class RuntimeContext:
     mode: str = "human"  # human | qa | sim
     responder: Any = None
     responder_meta: dict[str, Any] = field(default_factory=dict)
-    config: QAConfig = field(default_factory=QAConfig)
+    config: RuntimeConfig = field(default_factory=RuntimeConfig)
     event_logger: Optional[Callable[[dict[str, Any]], None]] = None
     sim_logger: Optional[Callable[[dict[str, Any]], None]] = None
     task_dir: Optional[Path] = None
@@ -68,15 +70,16 @@ class QAContext:
     rng: Any = None
 
 
-_CTX: contextvars.ContextVar[Optional[QAContext]] = contextvars.ContextVar("psyflow_qa_ctx", default=None)
+_CTX: contextvars.ContextVar[Optional[RuntimeContext]] = contextvars.ContextVar(
+    "psyflow_runtime_ctx", default=None
+)
 
 
-def get_context() -> Optional[QAContext]:
+def get_context() -> Optional[RuntimeContext]:
     return _CTX.get()
 
 
 def log_event(event: dict[str, Any]) -> None:
-    """Best-effort QA event logging (never raise)."""
     ctx = get_context()
     if ctx is None or ctx.event_logger is None:
         return
@@ -87,7 +90,6 @@ def log_event(event: dict[str, Any]) -> None:
 
 
 def log_sim_event(event: dict[str, Any]) -> None:
-    """Best-effort simulation event logging (never raise)."""
     ctx = get_context()
     if ctx is None or ctx.sim_logger is None:
         return
@@ -115,8 +117,7 @@ def _log(ev: dict[str, Any]) -> None:
 
 
 @contextlib.contextmanager
-def qa_context(ctx: QAContext):
-    """Activate QA context for the current execution context."""
+def runtime_context(ctx: RuntimeContext):
     token = _CTX.set(ctx)
     try:
         yield ctx
@@ -130,54 +131,38 @@ def qa_context(ctx: QAContext):
         _CTX.reset(token)
 
 
-def context_from_env(
+def context_from_config(
     *,
-    task_dir: str | os.PathLike | None = None,
+    task_dir: str | Path | None = None,
     config: dict[str, Any] | None = None,
-) -> QAContext:
-    """Build QA/sim context from environment variables (+ optional config mapping)."""
+    mode: str = "human",
+) -> RuntimeContext:
+    """Build runtime context from config with explicit mode selection."""
     raw_cfg = config
     if isinstance(config, dict) and isinstance(config.get("raw"), dict):
         raw_cfg = config.get("raw")
 
-    mode_cfg = str(_cfg_get(raw_cfg, ("sim", "mode"), "human") or "human").strip().lower()
-    mode = os.getenv("PSYFLOW_MODE", mode_cfg).strip().lower() or "human"
+    mode_cfg = _normalize_mode(_cfg_get(raw_cfg, ("sim", "mode"), mode))
+    mode = _normalize_mode(mode, default=mode_cfg)
 
     default_output_dir = "outputs/sim" if mode == "sim" else "outputs/qa"
     output_dir_cfg = _cfg_get(raw_cfg, ("sim", "output_dir"), None) or _cfg_get(raw_cfg, ("qa", "output_dir"), None)
-    output_dir = os.getenv("PSYFLOW_QA_OUTPUT_DIR", str(output_dir_cfg or default_output_dir))
-
-    enable_scaling = _as_bool(
-        os.getenv(
-            "PSYFLOW_QA_ENABLE_SCALING",
-            str(int(_as_bool(_cfg_get(raw_cfg, ("qa", "enable_scaling"), False)))),
-        ),
-        False,
-    )
-    timing_scale = float(os.getenv("PSYFLOW_QA_TIMING_SCALE", str(_cfg_get(raw_cfg, ("qa", "timing_scale"), 1.0))))
-    min_frames = int(os.getenv("PSYFLOW_QA_MIN_FRAMES", str(_cfg_get(raw_cfg, ("qa", "min_frames"), 2))))
-    strict = _as_bool(
-        os.getenv("PSYFLOW_QA_STRICT", str(int(_as_bool(_cfg_get(raw_cfg, ("qa", "strict"), False))))),
-        False,
-    )
-    max_wait_s = float(os.getenv("PSYFLOW_QA_MAX_WAIT_S", str(_cfg_get(raw_cfg, ("qa", "max_wait_s"), 10.0))))
+    output_dir = str(output_dir_cfg or default_output_dir)
 
-    sim_policy = str(
-        os.getenv(
-            "PSYFLOW_SIM_POLICY",
-            str(_cfg_get(raw_cfg, ("sim", "policy"), "strict" if strict else "warn")),
-        )
-    ).strip().lower()
+    enable_scaling = _as_bool(_cfg_get(raw_cfg, ("qa", "enable_scaling"), False), False)
+    timing_scale = float(_cfg_get(raw_cfg, ("qa", "timing_scale"), 1.0))
+    min_frames = int(_cfg_get(raw_cfg, ("qa", "min_frames"), 2))
+    strict = _as_bool(_cfg_get(raw_cfg, ("qa", "strict"), False), False)
+    max_wait_s = float(_cfg_get(raw_cfg, ("qa", "max_wait_s"), 10.0))
+
+    sim_policy = str(_cfg_get(raw_cfg, ("sim", "policy"), "strict" if strict else "warn")).strip().lower()
     if sim_policy not in ("strict", "warn", "coerce"):
         sim_policy = "strict" if strict else "warn"
 
-    default_rt_s = float(os.getenv("PSYFLOW_SIM_DEFAULT_RT_S", str(_cfg_get(raw_cfg, ("sim", "default_rt_s"), 0.2))))
-    clamp_rt = _as_bool(
-        os.getenv("PSYFLOW_SIM_CLAMP_RT", str(int(_as_bool(_cfg_get(raw_cfg, ("sim", "clamp_rt"), False))))),
-        False,
-    )
+    default_rt_s = float(_cfg_get(raw_cfg, ("sim", "default_rt_s"), 0.2))
+    clamp_rt = _as_bool(_cfg_get(raw_cfg, ("sim", "clamp_rt"), False), False)
 
-    cfg = QAConfig(
+    cfg = RuntimeConfig(
         enable_scaling=enable_scaling,
         timing_scale=timing_scale,
         min_frames=min_frames,
@@ -193,21 +178,15 @@ def context_from_env(
         or "unknown_task"
     )
     task_version = _cfg_get(raw_cfg, ("task", "task_version"), _cfg_get(config, ("task_config", "task_version"), None))
-    seed = int(os.getenv("PSYFLOW_SIM_SEED", str(_cfg_get(raw_cfg, ("sim", "seed"), 0))))
-    participant_id = str(
-        os.getenv(
-            "PSYFLOW_PARTICIPANT_ID",
-            _cfg_get(raw_cfg, ("sim", "participant_id"), _cfg_get(raw_cfg, ("task", "participant_id"), "p000")),
-        )
-        or "p000"
-    )
+    seed = int(_cfg_get(raw_cfg, ("sim", "seed"), 0))
+    participant_id = str(_cfg_get(raw_cfg, ("sim", "participant_id"), _cfg_get(raw_cfg, ("task", "participant_id"), "p000")) or "p000")
     default_session_id = f"{mode}-{participant_id}-seed{seed}"
-    session_id = str(os.getenv("PSYFLOW_SESSION_ID", str(_cfg_get(raw_cfg, ("sim", "session_id"), default_session_id))))
+    session_id = str(_cfg_get(raw_cfg, ("sim", "session_id"), default_session_id))
     session = SessionInfo(
         participant_id=participant_id,
         session_id=session_id,
         seed=seed,
-        mode=mode if mode in ("human", "qa", "sim") else "human",
+        mode=mode,
         task_name=task_name,
         task_version=task_version,
     )
@@ -219,9 +198,9 @@ def context_from_env(
     events_path = out / "qa_events.jsonl"
     event_logger = make_jsonl_logger(events_path) if mode in ("qa", "sim") else None
 
-    sim_log_default = "outputs/sim/sim_events.jsonl" if mode == "sim" else str(Path(output_dir) / "sim_events.jsonl")
+    sim_log_default = str(Path(output_dir) / "sim_events.jsonl")
     sim_log_cfg = _cfg_get(raw_cfg, ("sim", "log_path"), sim_log_default)
-    sim_log_path = os.getenv("PSYFLOW_SIM_LOG_PATH", str(sim_log_cfg))
+    sim_log_path = str(sim_log_cfg)
     sim_log = (tdir / sim_log_path) if (tdir is not None and not Path(sim_log_path).is_absolute()) else Path(sim_log_path)
     sim_logger = make_sim_jsonl_logger(sim_log) if mode in ("qa", "sim") else None
 
@@ -242,7 +221,7 @@ def context_from_env(
             allow_fallback=not strict,
         )
 
-    return QAContext(
+    return RuntimeContext(
         mode=mode,
         responder=responder,
         responder_meta=responder_meta,