diff --git a/api/controllers/console/workflow_copilot.py b/api/controllers/console/workflow_copilot.py
new file mode 100644
index 00000000000..89d463bf4bc
--- /dev/null
+++ b/api/controllers/console/workflow_copilot.py
@@ -0,0 +1,161 @@
+"""Console HTTP surface for the workflow copilot (P3c) — JSON routes.
+
+Flask-RESTX resources under /console/api/workflow-copilot/*, matching the
+retired Go enterprise transport byte-for-byte so cutover is a frontend
+base-path swap. Auth + CSRF come from the console guard stack (CSRF is enforced
+inside login_required for every method); RBAC/authz is enforced here at the
+boundary — the Celery advance task trusts the Actor. The route LOGIC lives in
+undecorated module functions (_create/_view/_action/_message) so it is unit
+testable without the Flask request stack; the Resource methods are thin
+auth-wrappers.
+
+Decorator injection contract (verified against controllers/console/wraps.py
+and the multi-decorator precedent in controllers/console/app/workflow_comment.py,
+e.g. ``def post(self, current_tenant_id, current_user, app_model, comment_id)``):
+``@with_current_user`` and ``@with_current_tenant_id`` do NOT inject their
+values as keyword arguments. Each calls its wrapped view as
+``view(self, injected_value, *args, **kwargs)`` — i.e. it PREPENDS the value
+as a positional argument immediately after ``self``. Stacking both means the
+decorator closer to the function (here, ``@with_current_tenant_id``, directly
+above ``@workflow_copilot_required``) contributes the positional argument
+closest to ``self``; the one above it (``@with_current_user``) contributes
+the next one. URL path parameters (e.g. ``session_id``) are untouched by
+either decorator and keep arriving as keyword arguments from Flask-RESTX's
+routing. ``workflow_copilot_required`` sits directly below
+``@with_current_tenant_id`` in the stack, so it receives ``current_tenant_id``
+positionally (as the argument right after ``self``), not as a keyword-only
+argument.
+"""
+
+import functools
+from collections.abc import Callable
+
+from flask import request
+from flask_restx import Resource
+
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+    with_current_tenant_id,
+    with_current_user,
+)
+from core.workflow_copilot.models import Action, Actor
+from libs.login import login_required
+from services.feature_service import FeatureService
+from services.workflow_copilot.wiring import build_service, copilot_error_response, session_view_to_dict
+
+from . import console_ns
+
+
+def workflow_copilot_required(func: Callable) -> Callable:
+    @functools.wraps(func)
+    def wrapper(self, current_tenant_id: str, *args, **kwargs):
+        if not FeatureService.get_features(current_tenant_id).workflow_copilot_enabled:
+            return {"code": "feature_unavailable"}, 403
+        return func(self, current_tenant_id, *args, **kwargs)
+
+    return wrapper
+
+
+def _actor(current_user, current_tenant_id: str) -> Actor:
+    return Actor(account_id=current_user.id, tenant_id=current_tenant_id)
+
+
+def _respond(fn: Callable[[], object]) -> tuple[dict, int]:
+    """Run a usecase call, serialize its SessionView, map known copilot errors
+    to (body, status). Unknown exceptions propagate to Flask's error handler."""
+    try:
+        return session_view_to_dict(fn()), 200
+    except Exception as exc:  # re-raised below if not a known copilot error
+        mapped = copilot_error_response(exc)
+        if mapped is None:
+            raise
+        return mapped
+
+
+def _create(body, actor: Actor) -> tuple[dict, int]:
+    if not isinstance(body, dict):
+        return {"code": "bad_request"}, 400
+    result, status = _respond(
+        lambda: build_service().create_fix_session(
+            app_id=body.get("app_id", ""),
+            actor=actor,
+            failed_run_id=body.get("failed_run_id") or None,
+            checklist_errors=body.get("checklist_errors") or None,
+        )
+    )
+    # create returns 201 on success; map-through keeps error statuses.
+    return result, (201 if status == 200 else status)
+
+
+def _view(session_id: str, actor: Actor) -> tuple[dict, int]:
+    return _respond(lambda: build_service().get_session_view(session_id, actor))
+
+
+def _action(session_id: str, body, actor: Actor) -> tuple[dict, int]:
+    if not isinstance(body, dict):
+        return {"code": "bad_request"}, 400
+    action = Action(
+        kind=body.get("kind", ""),
+        payload=body.get("payload") or {},
+        base_version=int(body.get("base_version", 0)),
+    )
+    return _respond(lambda: build_service().submit_action(session_id, actor, action))
+
+
+def _message(session_id: str, body, actor: Actor) -> tuple[dict, int]:
+    if not isinstance(body, dict):
+        return {"code": "bad_request"}, 400
+    return _respond(
+        lambda: build_service().submit_message(
+            session_id, actor, body.get("text", ""), int(body.get("base_version", 0))
+        )
+    )
+
+
+@console_ns.route("/workflow-copilot/sessions")
+class WorkflowCopilotSessionsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
+    @workflow_copilot_required
+    def post(self, current_tenant_id, current_user):
+        return _create(request.get_json(silent=True), _actor(current_user, current_tenant_id))
+
+
+@console_ns.route("/workflow-copilot/sessions/<string:session_id>")
+class WorkflowCopilotSessionApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
+    @workflow_copilot_required
+    def get(self, current_tenant_id, current_user, session_id):
+        return _view(session_id, _actor(current_user, current_tenant_id))
+
+
+@console_ns.route("/workflow-copilot/sessions/<string:session_id>/actions")
+class WorkflowCopilotActionsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
+    @workflow_copilot_required
+    def post(self, current_tenant_id, current_user, session_id):
+        return _action(session_id, request.get_json(silent=True), _actor(current_user, current_tenant_id))
+
+
+@console_ns.route("/workflow-copilot/sessions/<string:session_id>/messages")
+class WorkflowCopilotMessagesApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
+    @workflow_copilot_required
+    def post(self, current_tenant_id, current_user, session_id):
+        return _message(session_id, request.get_json(silent=True), _actor(current_user, current_tenant_id))
diff --git a/api/tests/unit_tests/controllers/console/test_workflow_copilot_routes.py b/api/tests/unit_tests/controllers/console/test_workflow_copilot_routes.py
new file mode 100644
index 00000000000..bdd62e381f1
--- /dev/null
+++ b/api/tests/unit_tests/controllers/console/test_workflow_copilot_routes.py
@@ -0,0 +1,187 @@
+"""Unit tests for the workflow copilot console JSON routes.
+
+These test the UNDECORATED module-level route-logic functions
+(``_create``/``_view``/``_action``/``_message``) and the ``workflow_copilot_required``
+gate directly -- NOT the decorated ``Resource`` methods. Calling a decorated
+resource method would trigger the real ``@login_required``/``@setup_required``
+auth stack, which needs a live Flask request context and a logged-in user;
+that is exercised by integration/e2e tests, not here.
+"""
+
+from unittest.mock import MagicMock
+
+import controllers.console.workflow_copilot as mod
+from core.workflow_copilot.errors import BusyError, ConflictError, NotFoundError
+from core.workflow_copilot.models import Actor
+from services.workflow_copilot.service import SessionView
+
+
+def _session_view(session_id: str = "s1") -> SessionView:
+    return SessionView(
+        session_id=session_id,
+        app_id="a1",
+        version=1,
+        state="fix.diagnose",
+        canvas_read_only=True,
+        run_status="executing",
+        interrupted=False,
+        conversation=[],
+    )
+
+
+def _account(account_id: str = "acc-1") -> MagicMock:
+    m = MagicMock()
+    m.id = account_id
+    return m
+
+
+def _actor() -> Actor:
+    return mod._actor(_account(), "ten-1")
+
+
+def test_actor_builds_from_current_user_and_tenant():
+    account = _account("acc-1")
+    actor = mod._actor(account, "ten-1")
+    assert actor == Actor(account_id="acc-1", tenant_id="ten-1")
+
+
+def test_create_returns_201_and_view(monkeypatch):
+    svc = MagicMock()
+    svc.create_fix_session.return_value = _session_view()
+    monkeypatch.setattr(mod, "build_service", lambda: svc)
+
+    actor = _actor()
+    body, status = mod._create({"app_id": "a1", "failed_run_id": "TR-1", "checklist_errors": []}, actor)
+
+    assert status == 201
+    assert body["session_id"] == "s1"
+    called = svc.create_fix_session.call_args
+    assert called.kwargs["app_id"] == "a1"
+    assert isinstance(called.kwargs["actor"], Actor)
+    assert called.kwargs["actor"] == actor
+
+
+def test_create_rejects_non_dict_body():
+    actor = _actor()
+    svc = MagicMock()
+    body, status = mod._create(None, actor)
+    assert (body, status) == ({"code": "bad_request"}, 400)
+    svc.create_fix_session.assert_not_called()
+
+
+def test_action_rejects_non_dict_body(monkeypatch):
+    svc = MagicMock()
+    monkeypatch.setattr(mod, "build_service", lambda: svc)
+    actor = _actor()
+
+    body, status = mod._action("s1", "notadict", actor)
+
+    assert (body, status) == ({"code": "bad_request"}, 400)
+    svc.submit_action.assert_not_called()
+
+
+def test_action_conflict_maps_409(monkeypatch):
+    svc = MagicMock()
+    svc.submit_action.side_effect = ConflictError("stale")
+    monkeypatch.setattr(mod, "build_service", lambda: svc)
+    actor = _actor()
+
+    body, status = mod._action("s1", {"kind": "run_verify", "payload": {}, "base_version": 1}, actor)
+
+    assert status == 409
+    assert body["code"] == "conflict"
+
+
+def test_action_busy_maps_409_session_busy(monkeypatch):
+    svc = MagicMock()
+    svc.submit_action.side_effect = BusyError("busy")
+    monkeypatch.setattr(mod, "build_service", lambda: svc)
+    actor = _actor()
+
+    body, status = mod._action("s1", {"kind": "publish", "payload": {}, "base_version": 2}, actor)
+
+    assert status == 409
+    assert body["code"] == "session_busy"
+
+
+def test_view_notfound_maps_generic_404(monkeypatch):
+    svc = MagicMock()
+    svc.get_session_view.side_effect = NotFoundError("session xyz not found")
+    monkeypatch.setattr(mod, "build_service", lambda: svc)
+    actor = _actor()
+
+    body, status = mod._view("s1", actor)
+
+    assert status == 404
+    assert body == {"code": "not_found"}
+    # the message text must NOT leak into the response
+    assert "xyz" not in str(body)
+
+
+def test_message_returns_200_and_view(monkeypatch):
+    svc = MagicMock()
+    svc.submit_message.return_value = _session_view()
+    monkeypatch.setattr(mod, "build_service", lambda: svc)
+    actor = _actor()
+
+    body, status = mod._message("s1", {"text": "hi", "base_version": 2}, actor)
+
+    assert status == 200
+    assert body["session_id"] == "s1"
+    svc.submit_message.assert_called_once_with("s1", actor, "hi", 2)
+
+
+def test_message_rejects_non_dict_body(monkeypatch):
+    svc = MagicMock()
+    monkeypatch.setattr(mod, "build_service", lambda: svc)
+    actor = _actor()
+
+    body, status = mod._message("s1", None, actor)
+
+    assert (body, status) == ({"code": "bad_request"}, 400)
+    svc.submit_message.assert_not_called()
+
+
+# --- workflow_copilot_required gate -----------------------------------------
+#
+# NOTE: the brief/dispatch assumed `@with_current_user`/`@with_current_tenant_id`
+# inject their values as KEYWORD arguments. Reading the real implementation in
+# controllers/console/wraps.py shows otherwise: both call
+# `view(self, injected_value, *args, **kwargs)`, i.e. they PREPEND the value as
+# a POSITIONAL argument immediately after `self`. This is confirmed by the
+# existing multi-decorator example in controllers/console/app/workflow_comment.py
+# (`def post(self, current_tenant_id, current_user, app_model, comment_id)`).
+# `workflow_copilot_required` sits directly below `@with_current_tenant_id`, so
+# it must accept `current_tenant_id` positionally (as the argument right after
+# `self`), not as a keyword-only argument. The gate is tested against that real
+# contract below.
+
+
+def test_workflow_copilot_required_blocks_when_feature_disabled(monkeypatch):
+    def dummy(_self, _current_tenant_id, *_args, **_kwargs):
+        return {"ok": True}, 200
+
+    wrapped = mod.workflow_copilot_required(dummy)
+
+    features = MagicMock()
+    features.workflow_copilot_enabled = False
+    monkeypatch.setattr(mod.FeatureService, "get_features", lambda _tenant_id: features)
+
+    result = wrapped(object(), "t")
+
+    assert result == ({"code": "feature_unavailable"}, 403)
+
+
+def test_workflow_copilot_required_passes_through_when_enabled(monkeypatch):
+    def dummy(_self, current_tenant_id, *_args, **_kwargs):
+        return {"ok": True, "current_tenant_id": current_tenant_id}, 200
+
+    wrapped = mod.workflow_copilot_required(dummy)
+
+    features = MagicMock()
+    features.workflow_copilot_enabled = True
+    monkeypatch.setattr(mod.FeatureService, "get_features", lambda _tenant_id: features)
+
+    result = wrapped(object(), "t")
+
+    assert result == ({"ok": True, "current_tenant_id": "t"}, 200)