feat(cli): sync workspace config and rerun artifacts

chipcompiler/cli/commands/workspace.py

@@ -18,8 +18,10 @@
     get_workspace_home,
     get_workspace_info,
     load_workspace,
+    refresh_workspace_config,
     run_workspace_flow,
     run_workspace_step,
+    sync_workspace_config,
 )
 
 workspace_app = typer.Typer(
@@ -237,6 +239,23 @@ def run_step_cmd(
     _dispatch_runtime(lambda: run_workspace_step(directory, step, rerun), json_output)
 
 
+@workspace_app.command("refresh-config", help="Refresh workspace config from parameters")
+def refresh_config_cmd(
+    directory: Annotated[str, typer.Option("--directory")] = "",
+    json_output: Annotated[bool, typer.Option("--json")] = False,
+) -> None:
+    _dispatch_runtime(lambda: refresh_workspace_config(directory), json_output)
+
+
+@workspace_app.command("sync-config", help="Sync managed workspace config fields to parameters")
+def sync_config_cmd(
+    directory: Annotated[str, typer.Option("--directory")] = "",
+    config_path: Annotated[str, typer.Option("--config-path")] = "",
+    json_output: Annotated[bool, typer.Option("--json")] = False,
+) -> None:
+    _dispatch_runtime(lambda: sync_workspace_config(directory, config_path), json_output)
+
+
 @workspace_app.command("get-info", help="Show workspace or step runtime information")
 def get_info_cmd(
     directory: Annotated[str, typer.Option("--directory")] = "",

chipcompiler/cli/workspace/service.py

@@ -106,7 +106,7 @@ def run_workspace_flow(directory: str, rerun: bool) -> dict:
     try:
         workspace, engine_flow = load_workspace_runtime(directory)
         if rerun:
-            engine_flow.clear_states()
+            _prepare_workspace_for_rerun(workspace, engine_flow)
         ok = engine_flow.run_steps(rerun=rerun)
         if not ok:
             return workspace_response(
@@ -154,6 +154,8 @@ def run_workspace_step(directory: str, step: str, rerun: bool) -> dict:
 
     try:
         workspace, engine_flow = load_workspace_runtime(directory)
+        if rerun:
+            _refresh_workspace_config(workspace)
         workspace_step = engine_flow.get_workspace_step(step)
         if workspace_step is None:
             state = engine_flow.run_step(step, rerun)
@@ -200,6 +202,114 @@ def run_workspace_step(directory: str, step: str, rerun: bool) -> dict:
     )
 
 
+def refresh_workspace_config(directory: str) -> dict:
+    cmd = "refresh_config"
+    response_data = {"directory": os.path.abspath(directory) if directory else "", "refreshed": False}
+    if not directory:
+        return workspace_response(
+            cmd,
+            "failed",
+            data=response_data,
+            message=["missing required field: directory"],
+        )
+
+    try:
+        workspace, _engine_flow = load_workspace_runtime(directory, create_step_workspaces=False)
+        resolved_directory = os.path.abspath(workspace.directory)
+        response_data["directory"] = resolved_directory
+
+        import chipcompiler.data as data_api
+
+        data_api.refresh_workspace_config(workspace)
+        response_data["refreshed"] = True
+    except WorkspaceValidationError as exc:
+        return workspace_response(cmd, "failed", data=response_data, message=[str(exc)])
+    except Exception as exc:
+        return workspace_response(
+            cmd,
+            "error",
+            data=response_data,
+            message=[f"refresh workspace config failed : {exc}"],
+        )
+
+    return workspace_response(
+        cmd,
+        "success",
+        data=response_data,
+        message=[f"refresh workspace config success : {response_data['directory']}"],
+    )
+
+
+def sync_workspace_config(directory: str, config_path: str) -> dict:
+    cmd = "sync_config"
+    response_data = {
+        "directory": os.path.abspath(directory) if directory else "",
+        "config_path": os.path.abspath(config_path) if config_path else "",
+        "parameters_changed": False,
+        "refreshed": False,
+    }
+    if not directory:
+        return workspace_response(
+            cmd,
+            "failed",
+            data=response_data,
+            message=["missing required field: directory"],
+        )
+    if not config_path:
+        return workspace_response(
+            cmd,
+            "failed",
+            data=response_data,
+            message=["missing required field: config_path"],
+        )
+
+    try:
+        workspace, _engine_flow = load_workspace_runtime(directory, create_step_workspaces=False)
+        resolved_directory = os.path.abspath(workspace.directory)
+        resolved_config_path = os.path.abspath(config_path)
+        response_data["directory"] = resolved_directory
+        response_data["config_path"] = resolved_config_path
+
+        config_dir = os.path.realpath(os.path.join(resolved_directory, "config"))
+        real_config_path = os.path.realpath(resolved_config_path)
+        if not _path_is_within(real_config_path, config_dir):
+            return workspace_response(
+                cmd,
+                "failed",
+                data=response_data,
+                message=[
+                    f"config path outside workspace config directory : {resolved_config_path}"
+                ],
+            )
+
+        import chipcompiler.data as data_api
+
+        parameters_changed = data_api.sync_workspace_config_to_parameters(
+            workspace,
+            resolved_config_path,
+        )
+        response_data["parameters_changed"] = parameters_changed
+        if parameters_changed:
+            data_api.refresh_workspace_config(workspace)
+            response_data["refreshed"] = True
+    except WorkspaceValidationError as exc:
+        return workspace_response(cmd, "failed", data=response_data, message=[str(exc)])
+    except Exception as exc:
+        return workspace_response(
+            cmd,
+            "error",
+            data=response_data,
+            message=[f"sync workspace config failed : {exc}"],
+        )
+
+    return workspace_response(
+        cmd,
+        "success",
+        data=response_data,
+        message=[f"sync workspace config success : {response_data['config_path']}"],
+    )
+
+
 def get_workspace_info(directory: str, step: str, info_id: str) -> dict:
     cmd = "get_info"
     step = step or ""
@@ -317,6 +427,25 @@ def build_flow_for_workspace(workspace, create_step_workspaces: bool = True):
     return engine_flow
 
 
+def _refresh_workspace_config(workspace):
+    import chipcompiler.data as data_api
+
+    data_api.refresh_workspace_config(workspace)
+
+
+def _prepare_workspace_for_rerun(workspace, engine_flow):
+    import chipcompiler.data as data_api
+
+    data_api.prepare_workspace_for_rerun(workspace, engine_flow)
+
+
+def _path_is_within(path: str, directory: str) -> bool:
+    try:
+        return os.path.commonpath([path, directory]) == directory
+    except ValueError:
+        return False
+
+
 def _init_db_engine_for_workspace_step(engine_flow, workspace_step):
     engine_db = getattr(engine_flow, "engine_db", None)
     if engine_db is None:

chipcompiler/data/__init__.py

@@ -18,6 +18,9 @@
     load_workspace,
     log_workspace,
     log_parameters,
+    prepare_workspace_for_rerun,
+    refresh_workspace_config,
+    sync_workspace_config_to_parameters,
     log_flow,
     log_workspace_step,
     update_step_config,
@@ -38,6 +41,9 @@
     'WorkspaceStep', 
     'build_workspace_config_paths',
     'init_workspace_config',
+    'prepare_workspace_for_rerun',
+    'refresh_workspace_config',
+    'sync_workspace_config_to_parameters',
     'update_step_config',
     'PDK',
     'get_pdk',