refactor(runtime): consolidate ecc workspace runtime

agent/engine.py

@@ -5,6 +5,7 @@
 
 from chipcompiler.data import StateEnum, WorkspaceStep
 from chipcompiler.engine.flow import EngineFlow
+from chipcompiler.engine.flow_completion import normalize_legacy_terminal_state
 from chipcompiler.engine.step_execution import get_process_rss_mb, track_current_process_memory
 from chipcompiler.utility.log import redirect_stdio_to_file
 
@@ -34,7 +35,7 @@ def run_step(self, workspace_step: WorkspaceStep | str, *, rerun: bool = False)
             self.clear_db_engine_after_step(workspace_step, StateEnum.Success)
             return StateEnum.Success
 
-        self._normalize_legacy_terminal_state(workspace_step, step_tag)
+        normalize_legacy_terminal_state(self, workspace_step, step_tag)
 
         start_time = time.time()
         timing_constraints = self.timing_constraint_facts()

chipcompiler/cli/command_handlers/param.py

@@ -11,6 +11,7 @@
     validate_pdk_target,
     validate_value,
 )
+from chipcompiler.rtl2gds import get_flow_builders, normalize_flow_step
 from chipcompiler.utility.file import write_text_atomic
 
 
@@ -29,6 +30,20 @@ def _manifest_mode_error(ctx: CommandContext) -> CommandResult | None:
     return None
 
 
+def _parameter_flow_error(schema, ctx: CommandContext) -> CommandResult | None:
+    preset = getattr(ctx.config, "flow_preset", "")
+    builder = get_flow_builders().get(preset)
+    if builder is None or schema.applies == "all" or schema.pdk_target is not None:
+        return None
+    flow_steps = {normalize_flow_step(step).casefold() for step, _tool, _state in builder()}
+    if normalize_flow_step(schema.applies).casefold() in flow_steps:
+        return None
+    return CommandResult.err(
+        [error_record("parameter_not_in_flow", param=schema.param, preset=preset)],
+        exit_code=1,
+    )
+
+
 def param_list(args, ctx: CommandContext) -> CommandResult:
     if getattr(args, "workspace", None) is not None:
         from chipcompiler.cli.command_handlers import workspace_params
@@ -45,12 +60,31 @@ def param_list(args, ctx: CommandContext) -> CommandResult:
     resolved, _ = resolve_parameters(toml_overrides=toml_overrides)
     project = ctx.project
 
-    selected_step = (getattr(args, "step", None) or "").casefold()
+    selected_step = normalize_flow_step(getattr(args, "step", None) or "").casefold()
     show_all = bool(getattr(args, "all", False))
+    preset = getattr(ctx.config, "flow_preset", "")
+    builder = get_flow_builders().get(preset)
+    flow_step_names = (
+        [normalize_flow_step(step).casefold() for step, _tool, _state in builder()]
+        if builder is not None
+        else []
+    )
+    flow_steps = set(flow_step_names)
+    first_flow_step = flow_step_names[0] if flow_step_names else ""
+    if selected_step and flow_steps and selected_step not in flow_steps:
+        return CommandResult.ok([])
     records = []
     for rp in resolved:
         s = rp.schema
-        if selected_step and selected_step not in {s.group.casefold(), s.applies.casefold()}:
+        applies = normalize_flow_step(s.applies).casefold()
+        if flow_steps and s.applies != "all" and applies not in flow_steps:
+            continue
+        schema_steps = {
+            normalize_flow_step(s.group).casefold(),
+            normalize_flow_step(s.applies).casefold(),
+        }
+        global_at_first_step = s.applies == "all" and selected_step == first_flow_step
+        if selected_step and selected_step not in schema_steps and not global_at_first_step:
             continue
         if not selected_step and not show_all and s.has_direct_target and not rp.is_explicit:
             continue
@@ -160,6 +194,9 @@ def param_set(args, ctx: CommandContext) -> CommandResult:
             ],
             exit_code=1,
         )
+    flow_error = _parameter_flow_error(schema, ctx)
+    if flow_error is not None:
+        return flow_error
 
     try:
         value = parse_value(raw_value, schema)
@@ -248,6 +285,9 @@ def param_unset(args, ctx: CommandContext) -> CommandResult:
             ],
             exit_code=1,
         )
+    flow_error = _parameter_flow_error(schema, ctx)
+    if flow_error is not None:
+        return flow_error
 
     config_path = _find_config_path(ctx.project_dir)
     if config_path is None:

chipcompiler/cli/command_handlers/project.py

@@ -444,6 +444,20 @@ def error(kind: str, **fields) -> CommandResult:
         return error("flow_range_requires_pair")
     if fresh_target and (command_input.resume or command_input.only is not None):
         return error("selector_requires_workspace")
+    if not fresh_target and ctx.project_state == "manifest" and cfg.manifest_driven:
+        return run_dispatch.dispatch_project_run(
+            command_input,
+            ctx,
+            cfg,
+            run_dir,
+            run_name,
+            cli_overrides,
+            flow_config,
+            project_state,
+            warning_records,
+            workspace_registered=workspace_registered,
+            execute_flow=execute_flow,
+        )
     errors = effective_config.validate_effective(
         ctx,
         cfg,

chipcompiler/cli/command_handlers/signoff.py

@@ -9,7 +9,7 @@ def inspect(command_input, ctx: CommandContext) -> CommandResult:
     if failure is not None:
         return failure
 
-    from chipcompiler.runtime.signoff_export import inspect_signoff_package
+    from chipcompiler.engine.signoff_export import inspect_signoff_package
 
     review = inspect_signoff_package(workspace)
     project = ctx.project
@@ -56,17 +56,18 @@ def export(command_input, ctx: CommandContext) -> CommandResult:
     if failure is not None:
         return failure
 
-    from chipcompiler.runtime.workspace_api import RuntimeApiError
-
     try:
-        from chipcompiler.runtime.signoff_export import export_signoff_package_archive
+        from chipcompiler.engine.signoff_export import (
+            SignoffExportError,
+            export_signoff_package_archive,
+        )
 
         output_path = export_signoff_package_archive(
             workspace,
             command_input.output_path,
             include_debug=command_input.include_debug,
         )
-    except RuntimeApiError as exc:
+    except SignoffExportError as exc:
         return CommandResult.err(
             [
                 error_record(

chipcompiler/cli/command_handlers/workspace_params.py

@@ -12,6 +12,7 @@
     workspace_param_step,
     workspace_param_value,
 )
+from chipcompiler.rtl2gds import normalize_flow_step
 
 
 def param_set(args, ctx: CommandContext) -> CommandResult:
@@ -62,15 +63,19 @@ def param_list(args, ctx: CommandContext) -> CommandResult:
     if workspace_error is not None:
         return workspace_error
     overrides = {record["key"] for record in workspace_param_diff(workspace)}
-    selected_step = (args.step or "").casefold()
+    selected_step = normalize_flow_step(args.step or "").casefold()
+    flow_steps = workspace.flow.data.get("steps", [])
+    first_step = normalize_flow_step(flow_steps[0]["name"]).casefold() if flow_steps else ""
     records = []
     for schema in list_schemas():
         if schema.pdk_target is not None or (not args.all and schema.param not in overrides):
             continue
-        if selected_step and selected_step not in {
-            schema.group.casefold(),
-            schema.applies.casefold(),
-        }:
+        schema_steps = {
+            normalize_flow_step(schema.group).casefold(),
+            normalize_flow_step(schema.applies).casefold(),
+        }
+        global_at_first_step = schema.applies == "all" and selected_step == first_step
+        if selected_step and selected_step not in schema_steps and not global_at_first_step:
             continue
         try:
             value = workspace_param_value(workspace, schema)
@@ -167,6 +172,8 @@ def _mutate(
             workspace, workspace_error = _load_workspace(ctx)
             if workspace_error is not None:
                 return workspace_error
+            if (Path(ctx.run_dir) / "home" / "engineering-snapshot.json").is_file():
+                return _mutate_via_engine(ctx, workspace, schema, requested_value, status)
             try:
                 result = mutation(workspace)
             except ValueError as exc:
@@ -223,6 +230,57 @@ def _mutate(
     return CommandResult.ok([record])
 
 
+def _mutate_via_engine(ctx, workspace, schema, requested_value: object, status: str):
+    from chipcompiler.data.workspace_parameters import workspace_param_diff, workspace_param_step
+    from chipcompiler.engine import update_workspace_step_configuration
+    from chipcompiler.engine.snapshot import read_engineering_snapshot
+    from chipcompiler.engine.workspace_lifecycle import WorkspaceLifecycleError
+
+    overrides = workspace_param_diff(workspace)
+    existing = next((item for item in overrides if item["key"] == schema.param), None)
+    if status == "unset" and existing is None:
+        return CommandResult.ok([_record(ctx, schema.param, None, "no_override")])
+    value = requested_value if status == "set" else existing["baseline"]
+    step = workspace_param_step(schema)
+    try:
+        revision = read_engineering_snapshot(workspace)["workspaceRevision"]
+        updated = update_workspace_step_configuration(
+            ctx.run_dir,
+            revision,
+            step,
+            {schema.param: value},
+            command_id="",
+        )
+    except WorkspaceLifecycleError as exc:
+        return CommandResult.err(
+            [error_record(exc.code, param=schema.param, reason=str(exc), **exc.details)]
+        )
+    except Exception as exc:
+        return CommandResult.err(
+            [error_record("workspace_param_refresh_failed", param=schema.param, reason=str(exc))]
+        )
+
+    updated_steps = updated.flow.steps()
+    target = normalize_flow_step(step).casefold()
+    start = next(
+        (
+            index
+            for index, item in enumerate(updated_steps)
+            if normalize_flow_step(item.get("name", "")).casefold() == target
+        ),
+        len(updated_steps),
+    )
+    invalidated = [
+        str(item.get("name", ""))
+        for item in updated_steps[start:]
+        if item.get("name") and item.get("state") == "Unstart"
+    ]
+    record = _record(ctx, schema.param, value, status)
+    record["from_step"] = step
+    record["invalidated_steps"] = invalidated
+    return CommandResult.ok([record])
+
+
 def _schema_and_workspace_error(key: str, ctx: CommandContext):
     schema = lookup_schema(key)
     if schema is None:

chipcompiler/cli/commands/report.py

@@ -31,7 +31,7 @@ def _finish(subcommand: str, command_input, handler) -> None:
     execute_command("report", command_input, handler, render_key=f"report:{subcommand}")
 
 
-@report_app.command("qor", help="Show the overall QoR score report (GUI scoring rules)")
+@report_app.command("qor", help="Show the overall QoR score report")
 def qor_cmd(
     *,
     output_path: OutputPathOption = None,

chipcompiler/cli/inspection/discovery.py

@@ -185,13 +185,20 @@ def resolve_command_workspace(workspace_arg, project, workspace_id, run_dir):
     instead. Returns (workspace, error-record-or-None); the caller maps a
     non-None record to a CommandResult.err.
     """
+    import inspect
+
     from chipcompiler.data import load_workspace
 
     path, error = resolve_workspace_path(workspace_arg, project, workspace_id, run_dir)
     if error is not None:
         return None, error
     try:
-        workspace = load_workspace(path)
+        load_kwargs = (
+            {"read_only": True}
+            if "read_only" in inspect.signature(load_workspace).parameters
+            else {}
+        )
+        workspace = load_workspace(path, **load_kwargs)
     except Exception as exc:
         return None, error_record("invalid_workspace", workspace=path, reason=str(exc))
     if workspace is None: