Write the source code for given component IDs to workspace files. Returns a compact JSON with file paths — no source code inline.
(
arguments: Dict[str, Any],
store: SessionStore,
)
| 19 | |
| 20 | |
| 21 | def handle_read_code_components( |
| 22 | arguments: Dict[str, Any], |
| 23 | store: SessionStore, |
| 24 | ) -> str: |
| 25 | """Write the source code for given component IDs to workspace files. |
| 26 | |
| 27 | Returns a compact JSON with file paths — no source code inline. |
| 28 | """ |
| 29 | session_id = arguments["session_id"] |
| 30 | session = store.get(session_id) |
| 31 | if session is None: |
| 32 | return json.dumps({"error": f"Session {session_id} not found or expired."}) |
| 33 | |
| 34 | if session.workspace is None: |
| 35 | return json.dumps({"error": "Session workspace not initialized."}) |
| 36 | |
| 37 | component_ids: List[str] = arguments["component_ids"] |
| 38 | components = session.components |
| 39 | workspace = session.workspace |
| 40 | |
| 41 | written_files: Dict[str, str] = {} # filename -> component_id |
| 42 | not_found: List[str] = [] |
| 43 | |
| 44 | for cid in component_ids: |
| 45 | node = components.get(cid) |
| 46 | if node is None: |
| 47 | not_found.append(cid) |
| 48 | continue |
| 49 | lang = getattr(node, "language", "") |
| 50 | source = (getattr(node, "source_code", None) or "").strip() |
| 51 | file_path = workspace.write_component_source(cid, source, lang) |
| 52 | written_files[file_path.name] = cid |
| 53 | |
| 54 | result = { |
| 55 | "written": len(written_files), |
| 56 | "not_found_count": len(not_found), |
| 57 | "not_found": not_found, |
| 58 | "source_dir": str(workspace.root / "sources"), |
| 59 | "files": written_files, |
| 60 | } |
| 61 | return json.dumps(result, indent=2, ensure_ascii=False) |