MultiAgenticSwarm VS Code Extension –

Ranjan Sapkota
September 2, 2025

1 Overview & Ground Rules

Context. We are in the middle of a full re-architecture: the old collaboration layer and file
I/O patterns are being replaced by a LangGraph-first approach with explicit ProjectState,
agent-as-subgraph, and a coordination prompt that compiles to an executable plan.

Your priorities (do these first).

• Library-first: MAS as a standalone Python library with a clean API. VS Code is a thin
UI on top.

• Agent-as-subgraph: An agent is a container of nodes (skills/tools). Toggling a tool =

toggling a node.

• State: A single ProjectState (blackboard) read/written by nodes; supports checkpoints

and replay.

• Coordination prompt: LLM emits a typed plan (tasks, edges, gates) that LangGraph
executes.

• Simulation-first: Deterministic mock provider and seeded tools before wiring frontier
LLMs.

Do not spend time on:

• Old collaboration code and ad-hoc triggers (being removed).

• Long bespoke “file messaging” patterns (moving to explicit state).

• Over-fitting the VS Code panel before the library is stable.

1
2 Critical Repairs & Minimal Viable Core
Two immediate blockers commonly seen:

1. Missing state module (multiagenticswarm/core/[Link]) despite docs referencing

it.

2. Broken API import in multiagenticswarm/api/[Link] (importing a non-existent

[Link]).

2.1 Implement ProjectState (single source of truth)

Create multiagenticswarm/core/[Link]. Keep it pure data + helpers. Serialize/deseri-
alize easily; store runs under runs/<timestamp>/[Link].
Listing 1: Skeleton of core/[Link]
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TypedDict, Dict, Any, List, Optional
from datetime import datetime

class Message(TypedDict, total=False):

role: str # "user" | "assistant" | "system" | "tool"
content: str
name: Optional[str]
timestamp: Optional[str]
tool_call_id: Optional[str]

class ToolResult(TypedDict, total=False):

id: str
name: str
input: Dict[str, Any]
output: Dict[str, Any]
success: bool
error: Optional[str]
started_at: str
finished_at: str
cost: Optional[float]
latency_ms: Optional[int]

class TaskRecord(TypedDict, total=False):

id: str
kind: str # "IMPLEMENT" | "TEST" | "CICD" | "DOCS" | ...
status: str # "PENDING" | "RUNNING" | "DONE" | "ERROR"
inputs: Dict[str, Any]
depends_on: List[str]

2
 created_at: str
updated_at: str
node_selector: Optional[str]
result_ref: Optional[str]
error: Optional[str]

@dataclass
class ProjectState:
project_id: str
repo_meta: Dict[str, Any] = field(default_factory=dict)
messages: List[Message] = field(default_factory=list)
tasks: Dict[str, TaskRecord] = field(default_factory=dict)
tool_calls: Dict[str, ToolResult] = field(default_factory=dict)
artifacts: Dict[str, str] = field(default_factory=dict) # path −> ref/hash
decisions: List[Dict[str, Any]] = field(default_factory=list)
checkpoints: List[str] = field(default_factory=list)
created_at: str = field(default_factory=lambda: [Link]().isoformat())
updated_at: str = field(default_factory=lambda: [Link]().isoformat())

def add_message(self, role: str, content: str, name: str | None = None) −> None:
[Link]({
"role": role, "content": content, "name": name,
"timestamp": [Link]().isoformat()
})
self.updated_at = [Link]().isoformat()

def add_task(self, rec: TaskRecord) −> None:

[Link][rec["id"]] = rec
self.updated_at = [Link]().isoformat()

# add: set_task_status, add_tool_result, add_artifact, snapshot(), restore(), to_dict(), from_dict()

Techniques. Keep writes centralized; add a simple checkpointer that dumps JSON after
each LangGraph node for replay/debug.

2.2 Replace the broken API import with a real FastAPI app
Create a proper create_app(System) and minimal endpoints. This boots locally and powers
the IDE later.
Listing 2: multiagenticswarm/api/[Link]
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from ..[Link] import System

3
def create_app(system: System) −> FastAPI:
app = FastAPI(title="MAS␣API", version="0.1.0")

class PlanRequest(BaseModel):
goal: str
constraints: dict | None = None

@[Link]("/health")
async def health():
return {"status": "ok"}

@[Link]("/v1/plan")
async def plan(req: PlanRequest):
try:
plan = await system.generate_plan([Link], [Link] or {})
return {"plan": plan}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

@[Link]("/v1/execute")
async def execute(payload: dict):
try:
result = await [Link](payload)
return {"result": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

@[Link]("/v1/state")
async def get_state():
return system.state_to_dict()

@[Link]("/v1/events")
async def events():
return getattr(system, "events", [])

return app

Add tests. Use [Link] to hit /health, /v1/plan, /v1/execute with the demo
orchestrator.

4
3 Agents, Nodes, Tools: Make the Model Real
3.1 Agent = container of nodes
We want node-level toggles (enable/disable). Each node maps to a tool/capability.
Listing 3: multiagenticswarm/core/[Link] (node-focused)
class AgentNode:
def __init__(self, name: str, tool_name: str, enabled: bool = True):
[Link] = name
self.tool_name = tool_name
[Link] = enabled

class Agent:
def __init__(self, name: str):
[Link] = name
[Link]: dict[str, AgentNode] = {}

def add_node(self, name: str, tool_name: str):

[Link][name] = AgentNode(name, tool_name, enabled=True)

def enable_node(self, name: str): [Link][name].enabled = True

def disable_node(self, name: str): [Link][name].enabled = False

def active_nodes(self) −> list[str]:

return [n for n, node in [Link]() if [Link]]

3.2 Tool Registry & IO validation

Strengthen ToolSpec and ToolExecutor: enforce JSON-schema, dry-run, caching, and policy
guards.
Listing 4: core/[Link] (spec idea)
from typing import Protocol, Any, Dict

class ToolSpec(dict): # plain dict for JSON−serializable spec

# keys: name, version, capabilities, input_schema, output_schema,
# side_effects: bool, idempotent: bool, timeout_ms: int
...

class Tool(Protocol):
spec: ToolSpec
async def run(self, tool_input: Dict[str, Any], ctx: Dict[str, Any]) −> Dict[str, Any]: ...
def validate_input(self, tool_input: Dict[str, Any]) −> None: ...
def validate_output(self, tool_output: Dict[str, Any]) −> None: ...
def preview(self, tool_input: Dict[str, Any]) −> Dict[str, Any]: ... # optional

5
 Listing 5: core/tool_executor.py (call path)
class ToolExecutor:
def __init__(self, registry, policy):
[Link] = registry # name −> Tool
[Link] = policy

async def call_tools(self, requests: list[dict], ctx: dict) −> list[dict]:

responses = []
for req in requests:
tool = [Link][req["name"]]
tool.validate_input(req["input"])
if [Link]("dryRun"):
proposed = getattr(tool, "preview", lambda x: {"preview": True})(req["input"])
[Link]({"id": req["id"], "name": req["name"], "result": proposed, "success": T
continue
with [Link](tool, ctx): # sandbox / allowlists
out = await [Link](req["input"], ctx)
tool.validate_output(out)
[Link]({"id": req["id"], "name": req["name"], "result": out, "success": True})
return responses

Techniques.
• Cache pure tools by (name, inputHash).
• Policy guard denies net/fs/shell unless node policy allows.
• Contract tests in tests/test_tool.py: valid IO, invalid IO, timeout path.

3.3 Mock LLM provider (deterministic simulation)

Before real models, ship a MockProvider so interns can run stable demos and tests.
Listing 6: llm/[Link] (mock example)
class LLMResponse(TypedDict, total=False):
text: str
tool_calls: list[dict]
reasoning: str | None

class LLMProvider(Protocol):
async def complete(self, messages: list[dict], ∗∗kwargs) −> LLMResponse: ...

class MockProvider:
async def complete(self, messages: list[dict], ∗∗kwargs) −> LLMResponse:
# Deterministic "plan" for demo
return {"text": "[MOCK]␣plan:␣analyze␣−>␣codegen␣−>␣unit_test␣−>␣docs␣−>␣ci"}

6
4 Orchestration Seam, Tests & CI
4.1 System orchestrator seam
Keep the orchestration pluggable: start with a demo orchestrator, later swap to LangGraph.
Listing 7: core/[Link] (thin orchestrator seam)
from .state import ProjectState

class System:
def __init__(self, config: dict, tool_executor, orchestrator_factory):
[Link] = config
[Link] = ProjectState(project_id=[Link]("project_id", "local"))
self.tool_executor = tool_executor
[Link] = orchestrator_factory([Link], tool_executor, config)

async def generate_plan(self, goal: str, constraints: dict):

return await [Link](goal, constraints)

async def execute(self, payload: dict):

return await [Link](payload)

def state_to_dict(self) −> dict:

# serialize ProjectState
return {
"project_id": [Link].project_id,
"tasks": [Link],
"artifacts": [Link],
"decisions": [Link],
}

4.2 Example plan JSON (what the coordination prompt should

emit)

Listing 8: Typed plan emitted by the coordination LLM

{
"goals": ["Create a notes app with login"],
"constraints": {"lang": "flutter", "coverageTarget": 0.75},
"tasks": [
{"id": "T1", "kind": "ANALYZE", "inputs": {"brief": "notes app"}, "dependsOn": []}
,
{"id": "T2", "kind": "CODEGEN", "inputs": {"feature": "auth+notes"}, "dependsOn":
["T1"]},
{"id": "T3", "kind": "TEST_UNIT", "inputs": {"suite": "core"}, "dependsOn": ["T2"]
},

7
 {"id": "T4", "kind": "DOCS", "inputs": {"audience": "dev"}, "dependsOn": ["T2"]},
{"id": "T5", "kind": "CICD", "inputs": {"provider": "github"}, "dependsOn": ["T3",
"T4"]}
],
"edges": [
{"from": "T3", "on_pass": "T4", "on_fail": "T2"},
{"from": "T4", "on_pass": "T5", "on_fail": "T2"}
],
"budgets": {"timeMin": 20,"tokenMax": 300000},
"toolPolicy": {"dryRunDefault": true}
}

4.3 Tests to add now

• State tests: serialization, checkpoint/restore, event append.

• Tool tests: valid/invalid IO; timeout; side-effect sandbox.

• API tests: /health, /v1/plan, /v1/execute via [Link].

• E2E demo: “notes app” with MockProvider; assert [Link], [Link], .github/workflows/ci.y

4.4 CI: build, test, and artifact

Add GitHub Actions to run tests and publish artifacts.
Listing 9: .github/workflows/[Link]
name: ci
on: [push, pull_request]
jobs:
test:
runs−on: ubuntu−latest
steps:
− uses: actions/checkout@v4
− uses: actions/setup−python@v5
with: { python−version: "3.11" }
− run: pipx install poetry
− run: poetry install −−no−interaction −−no−root
− run: poetry run pytest −q −−maxfail=1
− run: poetry run black −−check .
− run: poetry run ruff check .
− run: poetry run mypy multiagenticswarm
(For the VS Code extension repo, add a separate workflow that runs npm ci, npm run
compile, and npx vsce package; upload the .vsix as an artifact.)

8
5 Productization Plan (MVP → GA)
5.1 Define the MVP (3–4 weeks)
Scope:

• MAS library compiles coordination prompt → plan JSON → LangGraph execu-

tion.

• Agent-as-subgraph with node toggles; Tool Registry with IO validation; Mock + one real
LLM provider.

• FastAPI endpoints: /health, /v1/plan, /v1/execute, /v1/state, /v1/events.

• Deterministic simulation mode (mock provider + seeded tools).

• One exemplary workflow: “notes app” (code, tests, CI, docs) with dry-run → apply.

• Evaluation harness: MAS vs. single long-context baseline on 20–30 internal tasks; log
success, time, cost, edit distance.

5.2 IDE integration (thin VS Code)

• The panel calls the FastAPI; it shows Agents, Node toggles, Plan, Activity, Ap-
provals.

• Enforce strict CSP (no inline JS); validate webview messages against a schema.

• CI: matrix build and .vsix artifact on each PR and tag.

5.3 Operational must-haves

• Observability: structured logs with traceId; event streams persisted per run; (optional)
OpenTelemetry later.

• Cost/budget: per-task limits (time, tokens); early stopping; retry with backoff; “debate
only when necessary.”

• Policy guardrails: dry-run by default; deny network/filesystem unless approved; redact

secrets in logs.

• Docs: Getting Started, Architecture, Tool Developer Guide, API Reference, Evaluation
Harness, Security.

• Licensing/Release: MIT for core; TestPyPI → PyPI; semantic versioning; CHANGELOG.

9
5.4 Evidence & evaluation
• 50–100 internal tasks with oracles (tests/CI or blinded rubric).

• Compare MAS vs. single long-context baseline under identical budgets.

• Report: success rate, integration quality (lint/typing/security/coverage), time, cost, hu-

man edit distance.

• Ablations: no-tests, no-tools, no-critique, static plan.

5.5 Three-sprint rollout

Sprint 1 (library hardening): implement [Link], fix API, MockProvider, pass tests,
run demo, publish v0.1.0-rc1 to TestPyPI.
Sprint 2 (LangGraph + real LLM): map 3–5 nodes (Analyze, Codegen, UnitTest, Docs,
CI), add checkpoints, run A/B on 30–50 tasks.
Sprint 3 (VS Code thin UI + polish): node toggles, activity, approvals; CSP; CI with
.vsix; publish v0.1.0 and traces.

5.6 Risks & mitigations

• LangGraph drift: keep MAS a thin layer (plan schema, node registry, state).

• LLM variance: seeded mock for demos; record seeds; bounded retries.

• Security: dry-run, deny risky tools by default; secrets scanning in CI.

• Cost blow-ups: strict budgets, early stop, confidence-gated critique.

6 Intern Checklist (Hand-off)

1. Create core/[Link]; add tests for serialization and checkpoints.

2. Replace API import; implement create_app(System); test with [Link].

3. Refactor Agent to manage nodes; add enable_node/disable_node.

4. Harden ToolExecutor: IO validation, policy guard, caching, deterministic previews.

5. Add MockProvider; make simulation mode default in tests.

6. Introduce system orchestrator seam; stub LangGraphOrchestrator; keep DemoOrchestra-

tor now.

7. Delete committed venvs (test_env/); update .gitignore.

8. Expand tests: state, tools, API, E2E demo; add coverage target in CI.

10
 9. Produce first deterministic “notes app” run: [Link], [Link], CI file.

10. Only then wire LangGraph nodes and edges, keeping the same plan/state contracts.

11