U-043: console deployment status view (v_deployment_status)

Adds src/deployment_query.py (list_deployments/get_current_deployment/ get_deployment -- no project_id filter needed, deployment has no project_id column since there's exactly one graphlemon self-target) + three new backend_server.py routes (GET /api/deployments, /current, /:id, the last with recent deploy_health_signal history) + a new v_deployment_status page in frontend/app.js polling every 5s (deployment.status changes don't go through c_ws_subscribe -- Deploy Supervisor writes deployment/ deploy_health_signal directly, no execution_event of its own). No 02-V spec exists for this view (PH5 was added after 02-V was written) -- DOM/class conventions follow the existing pages' own patterns. Verified against a real running server via urllib (browser extension unavailable this session), plus 9 deployment_query unit tests and 5 real HTTP route tests. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-043: console deployment status view (v_deployment_status)
Adds src/deployment_query.py (list_deployments/get_current_deployment/ get_deployment -- no project_id filter needed, deployment has no project_id column since there's exactly one graphlemon self-target) + three new backend_server.py routes (GET /api/deployments, /current, /:id, the last with recent deploy_health_signal history) + a new v_deployment_status page in frontend/app.js polling every 5s (deployment.status changes don't go through c_ws_subscribe -- Deploy Supervisor writes deployment/ deploy_health_signal directly, no execution_event of its own). No 02-V spec exists for this view (PH5 was added after 02-V was written) -- DOM/class conventions follow the existing pages' own patterns. Verified against a real running server via urllib (browser extension unavailable this session), plus 9 deployment_query unit tests and 5 real HTTP route tests. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 05292668
Showing 8 changed files with 408 additions and 2 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
frontend/app.js
frontend/index.html
src/backend_server.py
src/deployment_query.py
src/test_backend_server.py
src/test_deployment_query.py
@@ -406,6 +406,9 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-043 [Value Path]** 控制台展示 `deployment` 状态与健康信号（02-V 新增视图）
   - 依赖：U-024, U-040
   - Done：能在控制台看到当前是否有 deployment 在进行中、处于状态机的哪个状态、最近一次健康信号采样结果
+  - **实现补充（2026-08-02）**：02-V 从没写过这个视图（PH5 是后加的 Phase，02-V 原文只覆盖 PH0-PH4），DOM/class 命名沿用现有页面（`v_skill_library` 等）的既有约定，没有 wireframe 可以照抄。新增 `src/deployment_query.py`（`list_deployments`/`get_current_deployment`/`get_deployment`，`deployment` 表没有 `project_id`——graphlemon 自己只有一份 self-target，不是每项目一份，所以读接口不需要 project_id 过滤，和 `v_skill_library` 的"平台级资产"是同一类情况）+ `backend_server.py` 三个新路由（`GET /api/deployments`、`GET /api/deployments/current`、`GET /api/deployments/:id`，后者带最近 50 条 `deploy_health_signal`）。前端 `v_deployment_status`：`deployment.status` 变化不走 `c_ws_subscribe`（Deploy Supervisor 直接写 `deployment`/`deploy_health_signal` 表，不产生自己的 `execution_event`），页面改用定时轮询（5s 间隔）而不是复用 WS 连接，路由切换时清理 interval 的方式和其余页面清理 WS 连接是同一个 `registerRouteCleanup` 约定。验证方式：9 个 `deployment_query` 单测 + 5 个真实 HTTP 路由测试（`test_backend_server.py` 新增 `DeploymentRouteTest`）+ 对着真实跑起来的服务器用 `urllib` 走了一遍全部三条路由确认真实响应结构正确（浏览器扩展本次不可用，`node --check` 过语法）。
+  - **提交时机的一个插曲，记录一下**：新增 `deployment_query.py` 后，`supervisor/test_deploy_cutover.py`/`test_deploy_rollback.py`/`test_deploy_stage.py` 里那些用 `git stash create` 模拟"对着当前未提交工作区做真实 candidate 拉起"的测试集体失败——根因是 `git stash create` 不会捕获**未跟踪的新文件**（这条限制 U-037 当时已经在自己的测试注释里写明过），`deployment_query.py` 是全新文件，没被 stash 快照捕捉到，检出出来的候选进程 `import` 它时直接失败，"候选进程起不来"。这不是这次改动本身有 bug（真正提交之后，`HEAD` 天然包含这份新文件，同一批测试会恢复通过），是这批测试自己的固有前提（"只覆盖已跟踪文件的改动"）第一次被一次"新增文件"的改动踩到；提交后立即重跑过一次完整套件确认确实恢复全绿，不是只凭猜测判断"这样应该没事"。
+  - 覆盖：`src/deployment_query.py`；`backend_server.py` 新增 `GET /api/deployments`/`GET /api/deployments/current`/`GET /api/deployments/:id`；`frontend/app.js`/`index.html` 新增 `v_deployment_status`
 ### PH5 已知交叉缺口（不阻塞设计定稿，实现前需要留意）
@@ -60,7 +60,7 @@
     {"id": "U-040", "phase": "PH5", "sub_phase": "PH5-S3", "name": "c_deploy_supervisor_monitor（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-039"], "coverage": {"model": ["deploy_health_signal"], "view": [], "controller": ["c_deploy_supervisor_monitor"], "process": [], "requirement": []}, "done": "四类信号采样正常（supervisor/deploy_monitor.py）；validator_approval_rate 偏离基线阈值时触发 anomaly_detected；approve 占比算法基于对 scheduler.finish_task_lifecycle 真实代码逻辑的核实（PASS 从不产生 bug_record，FAIL 必然产生），非字面 SDD 定义"},
     {"id": "U-041", "phase": "PH5", "sub_phase": "PH5-S3", "name": "c_deploy_supervisor_rollback（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-038", "U-040"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_rollback"], "process": [], "requirement": []}, "done": "回滚路径 A/B 各跑通一次（supervisor/deploy_rollback.py，真实进程+真实 git 仓库验证）；旧版本也起不来时正确终态到 blocked_pending_human（真实场景验证，非假设）；git revert 冲突有明确处理（真实构造冲突历史验证过，含 --abort 清理）"},
     {"id": "U-042", "phase": "PH5", "sub_phase": "PH5-S3", "name": "自举闭环端到端验收（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-035", "U-041"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "真实复现 self-development-feasibility.md 9 步场景，全程无需人工评审代码内容（supervisor/self_hosting_e2e.py 编排层已建好）；真实双实例 E2E 用例（RUN_REAL_SELF_HOSTING_E2E=1，独立于常规 RUN_REAL_AGENT_E2E 的单独环境变量）默认跳过，需显式环境变量开启且需额外用户确认才真正执行；日常回归只跑 mock 版本（6 个编排逻辑测试，已跑通）"},
-    {"id": "U-043", "phase": "PH5", "sub_phase": "PH5-S4", "name": "控制台展示 deployment 状态与健康信号", "classification": "Value Path", "depends_on": ["U-024", "U-040"], "coverage": {"model": [], "view": ["v_deployment_status"], "controller": [], "process": [], "requirement": []}, "done": "能看到当前 deployment 状态机所处状态与最近一次健康信号采样结果"}
+    {"id": "U-043", "phase": "PH5", "sub_phase": "PH5-S4", "name": "控制台展示 deployment 状态与健康信号（见 development-plan.md 2026-08-02 实现补充）", "classification": "Value Path", "depends_on": ["U-024", "U-040"], "coverage": {"model": [], "view": ["v_deployment_status"], "controller": ["c_deployment_list", "c_deployment_current", "c_deployment_get"], "process": [], "requirement": []}, "done": "能看到当前 deployment 状态机所处状态与最近一次健康信号采样结果；02-V 从未定义过这个视图（PH5 后加），DOM/class 约定沿用现有页面；真实服务器 + urllib 验证过全部三条新路由"}
   ],
   "classification_summary": {
     "Value Path": 17,
@@ -79,6 +79,7 @@ const routes = {
   "#/tasks": { render: renderTaskBoard, mount: mountTaskBoard },
   "#/artifacts": { render: renderArtifactBrowser, mount: mountArtifactBrowser },
   "#/skills": { render: renderSkillLibrary, mount: mountSkillLibrary },
+  "#/deployment": { render: renderDeploymentStatus, mount: mountDeploymentStatus },
   "#/settings/concurrency": { render: renderConcurrencySettings, mount: mountConcurrencySettings },
   "#/settings/budget": { render: renderBudgetSettings, mount: mountBudgetSettings },
   "#/settings/log-retention": { render: renderLogRetentionSettings, mount: mountLogRetentionSettings },
@@ -1248,6 +1249,117 @@ function mountSkillAuditLog(skillId) {
   refresh();
 }
+// ---- v_deployment_status (U-043, PH5 -- no 02-V spec exists for this view,
+// it's a PH5-only addition; DOM/class conventions below follow the existing
+// pages' own patterns rather than a wireframe that was never written) ----
+// Not project-scoped (there's exactly one graphlemon self-target, not one
+// per project, same reasoning v_skill_library's pages already apply).
+// deployment.status changes aren't pushed over c_ws_subscribe (Deploy
+// Supervisor writes deployment/deploy_health_signal rows directly, no
+// execution_event of its own) -- this page polls on an interval instead of
+// reusing connectEventSocket, and cleans the interval up via
+// registerRouteCleanup like every other page cleans up its own sockets.
+
+const DEPLOYMENT_POLL_INTERVAL_MS = 5000;
+
+const DEPLOYMENT_STATUS_LABELS = {
+  staged: "已暂存", candidate_starting: "候选启动中", health_check_running: "健康检查中",
+  health_check_passed: "健康检查通过", health_check_failed: "健康检查失败",
+  cutover_in_progress: "切换中", live: "已上线", monitoring: "监控中",
+  anomaly_detected: "检测到异常", rollback_in_progress: "回滚中", rolled_back: "已回滚",
+  rollback_failed: "回滚失败", blocked_pending_human: "需要人工介入",
+  monitoring_window_clean: "监控窗口正常", finalized: "已完成",
+};
+
+function renderDeploymentStatus() {
+  return `
+    <div class="page page-wide">
+      <h1>部署状态</h1>
+      <div id="deployment_current"><div class="hint">加载中…</div></div>
+      <h2 style="margin-top:24px;">历史部署</h2>
+      <div id="deployment_history"><div class="hint">加载中…</div></div>
+    </div>
+  `;
+}
+
+function mountDeploymentStatus() {
+  const currentEl = document.getElementById("deployment_current");
+  const historyEl = document.getElementById("deployment_history");
+
+  function renderSignals(signals) {
+    if (!signals || signals.length === 0) return '<div class="hint">尚无健康信号采样</div>';
+    return '<div class="signal-list">' + signals.map((s) =>
+      '<div class="signal-row status-' + escapeHtml(s.status) + '">'
+      + '<span class="type">' + escapeHtml(s.signal_type) + '</span>'
+      + '<span class="tag">' + escapeHtml(s.status) + '</span>'
+      + '<span class="t">' + escapeHtml(s.sampled_at) + '</span>'
+      + (s.detail ? '<span class="detail">' + escapeHtml(s.detail) + '</span>' : '')
+      + '</div>'
+    ).join('') + '</div>';
+  }
+
+  function renderCurrent(current) {
+    if (!current) {
+      currentEl.innerHTML = '<div class="panel"><div class="hint">当前没有进行中的 deployment</div></div>';
+      return;
+    }
+    const label = DEPLOYMENT_STATUS_LABELS[current.status] || current.status;
+    currentEl.innerHTML = (
+      '<div class="panel">'
+      + '<div class="top"><b>deployment_id: ' + escapeHtml(current.deployment_id) + '</b>'
+      + '<span class="tag status-' + escapeHtml(current.status) + '">' + escapeHtml(label) + '</span></div>'
+      + '<div class="summary">commit ' + escapeHtml((current.git_commit_hash || '').slice(0, 12))
+      + ' · started_at ' + escapeHtml(current.started_at || '-') + '</div>'
+      + '</div>'
+    );
+  }
+
+  function renderHistory(deployments) {
+    if (deployments.length === 0) {
+      historyEl.innerHTML = '<div class="hint">暂无历史部署记录</div>';
+      return;
+    }
+    historyEl.innerHTML = deployments.map((d) => {
+      const label = DEPLOYMENT_STATUS_LABELS[d.status] || d.status;
+      return (
+        '<div class="deployment-row" data-deployment-id="' + d.deployment_id + '">'
+        + '<span class="name">' + escapeHtml(d.deployment_id.slice(0, 8)) + '</span>'
+        + '<span class="tag status-' + escapeHtml(d.status) + '">' + escapeHtml(label) + '</span>'
+        + '<span class="t">' + escapeHtml(d.cutover_at || d.started_at || '-') + '</span>'
+        + '</div>'
+      );
+    }).join('');
+
+    for (const row of historyEl.querySelectorAll('.deployment-row')) {
+      row.addEventListener('click', async () => {
+        const detail = await fetch('/api/deployments/' + encodeURIComponent(row.dataset.deploymentId)).then((r) => r.json());
+        currentEl.innerHTML = (
+          '<div class="panel">'
+          + '<div class="top"><b>deployment_id: ' + escapeHtml(detail.deployment_id) + '</b>'
+          + '<span class="tag status-' + escapeHtml(detail.status) + '">'
+          + escapeHtml(DEPLOYMENT_STATUS_LABELS[detail.status] || detail.status) + '</span></div>'
+          + '<div class="summary">commit ' + escapeHtml((detail.git_commit_hash || '').slice(0, 12)) + '</div>'
+          + '<h3>最近健康信号</h3>' + renderSignals(detail.health_signals)
+          + '</div>'
+        );
+      });
+    }
+  }
+
+  async function refresh() {
+    const [current, history] = await Promise.all([
+      fetch('/api/deployments/current').then((r) => r.json()),
+      fetch('/api/deployments').then((r) => r.json()),
+    ]);
+    renderCurrent(current);
+    renderHistory(history);
+  }
+
+  refresh();
+  const intervalId = setInterval(refresh, DEPLOYMENT_POLL_INTERVAL_MS);
+  registerRouteCleanup(() => clearInterval(intervalId));
+}
+
 // ---- v_concurrency_settings / v_budget_settings / v_log_retention_settings
 // (02-V §2.11/§2.12/§2.13) -- three small config pages, none of them
 // project-scoped (all three tables are platform-wide singletons, config_id
@@ -11,6 +11,7 @@
   <a href="#/tasks">Tasks</a>
   <a href="#/artifacts">Artifacts</a>
   <a href="#/skills">Skills</a>
+  <a href="#/deployment">Deployment</a>
   <a href="#/settings/concurrency">Settings</a>
 </nav>
 <main id="app"></main>
@@ -37,7 +37,7 @@ import urllib.parse
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from src import (
-    artifact_query, budget_guardrail, bug_record, execution_log,
+    artifact_query, budget_guardrail, bug_record, deployment_query, execution_log,
     org_loop_route, paths, project_scaffold, schema, settings_config,
     skill_query, skill_update, task_create, task_query, ws_subscribe,
 )
@@ -127,6 +127,7 @@ _WS_WORK_NODES_PREFIX = &quot;/api/ws/work-nodes/&quot;
 _LOG_SUFFIX = "/log"
 _ARTIFACTS_PREFIX = "/api/artifacts/"
 _SKILLS_PREFIX = "/api/skills/"
+_DEPLOYMENTS_PREFIX = "/api/deployments/"
 class _Handler(BaseHTTPRequestHandler):
@@ -181,6 +182,17 @@ class _Handler(BaseHTTPRequestHandler):
             skill_id = parsed.path[len(_SKILLS_PREFIX):-len("/audit")]
             self._handle_skill_audit_log(skill_id)
             return
+        if parsed.path == "/api/deployments":
+            self._handle_deployment_list()
+            return
+        if parsed.path == "/api/deployments/current":
+            self._handle_deployment_current()
+            return
+        if parsed.path.startswith(_DEPLOYMENTS_PREFIX):
+            deployment_id = parsed.path[len(_DEPLOYMENTS_PREFIX):]
+            if deployment_id and "/" not in deployment_id:
+                self._handle_deployment_get(deployment_id)
+                return
         entry = _STATIC_FILES.get(self.path)
         if entry is None:
@@ -448,6 +460,25 @@ class _Handler(BaseHTTPRequestHandler):
         else:
             self._send_json(200, result)
+    def _handle_deployment_list(self):
+        with self.server.conn_lock:
+            result = deployment_query.list_deployments(self.server.conn)
+        self._send_json(200, result)
+
+    def _handle_deployment_current(self):
+        with self.server.conn_lock:
+            result = deployment_query.get_current_deployment(self.server.conn)
+        self._send_json(200, result)
+
+    def _handle_deployment_get(self, deployment_id):
+        try:
+            with self.server.conn_lock:
+                result = deployment_query.get_deployment(self.server.conn, deployment_id)
+        except deployment_query.DeploymentNotFoundError:
+            self._send_json(404, {"error": "NOT_FOUND"})
+        else:
+            self._send_json(200, result)
+
     def _handle_skill_list(self):
         with self.server.conn_lock:
             result = skill_query.list_skills(self.server.conn)
+"""U-043: read-side controllers for `v_deployment_status` -- no operation_id
+for these exists anywhere in 03-C/the MVCP JSON (this whole view is a PH5
+addition with no 02-V spec to begin with, see development-plan.md's own
+U-043 note), same "view's own filter/display needs directly justify a new
+read endpoint" precedent as `c_task_list`/`c_artifact_list` before it.
+`deployment` has no `project_id` column (01-M 1.15) -- there's exactly one
+graphlemon self-target, not one per project, so these never take a
+project_id filter the way task/artifact listing does.
+"""
+
+_STATUSES = (
+    "staged", "candidate_starting", "health_check_running", "health_check_passed",
+    "health_check_failed", "cutover_in_progress", "live", "monitoring",
+    "anomaly_detected", "rollback_in_progress", "rolled_back", "rollback_failed",
+    "blocked_pending_human", "monitoring_window_clean", "finalized",
+)
+_TERMINAL_STATUSES = {"rolled_back", "rollback_failed", "finalized", "blocked_pending_human"}
+
+
+class DeploymentNotFoundError(Exception):
+    """NOT_FOUND -- c_deployment_get on a nonexistent deployment_id."""
+
+
+def _deployment_row_to_dict(r):
+    return {
+        "deployment_id": r[0], "git_commit_hash": r[1], "previous_commit_hash": r[2], "status": r[3],
+        "candidate_port": r[4], "candidate_instance_dir": r[5],
+        "started_at": r[6], "cutover_at": r[7], "monitoring_closed_at": r[8],
+        "rollback_at": r[9], "rollback_reason": r[10], "previous_instance_dir": r[11],
+        "is_active": r[3] not in _TERMINAL_STATUSES,
+    }
+
+
+_DEPLOYMENT_COLUMNS = (
+    "deployment_id, git_commit_hash, previous_commit_hash, status, candidate_port, candidate_instance_dir, "
+    "started_at, cutover_at, monitoring_closed_at, rollback_at, rollback_reason, previous_instance_dir"
+)
+
+
+def list_deployments(conn, limit=20):
+    """Newest-first (`rowid DESC` -- `deployment` has no created_at column
+    of its own, `started_at` is nullable until c_deploy_supervisor_stage
+    actually sets it, same ordering convention U-016/U-037 already use
+    elsewhere for "most recent" when a table lacks a dedicated timestamp)."""
+    rows = conn.execute(
+        "SELECT {} FROM deployment ORDER BY rowid DESC LIMIT ?".format(_DEPLOYMENT_COLUMNS), (limit,)
+    ).fetchall()
+    return [_deployment_row_to_dict(r) for r in rows]
+
+
+def get_current_deployment(conn):
+    """The single non-terminal deployment, if any (01-M 1.15's own
+    `ux_deployment_active` constraint guarantees at most one) -- what
+    `v_deployment_status` actually wants front and center: "当前是否有
+    deployment 在进行中". Returns None when nothing is active (the common
+    case -- most of the time nothing is deploying)."""
+    row = conn.execute(
+        "SELECT {} FROM deployment WHERE status NOT IN ({}) ORDER BY rowid DESC LIMIT 1".format(
+            _DEPLOYMENT_COLUMNS, ", ".join("'{}'".format(s) for s in _TERMINAL_STATUSES)
+        )
+    ).fetchone()
+    return _deployment_row_to_dict(row) if row else None
+
+
+def get_deployment(conn, deployment_id):
+    """Includes recent `deploy_health_signal` rows (newest first) -- "最近
+    一次健康信号采样结果" per this Unit's own Done line, but the full recent
+    history (not just the latest) is what actually lets a human tell "one
+    blip" from "a real pattern" at a glance, same reasoning U-026's
+    `v_graph_visualizer` tooltip already applies to per-node detail."""
+    row = conn.execute(
+        "SELECT {} FROM deployment WHERE deployment_id = ?".format(_DEPLOYMENT_COLUMNS), (deployment_id,)
+    ).fetchone()
+    if row is None:
+        raise DeploymentNotFoundError(deployment_id)
+    result = _deployment_row_to_dict(row)
+    result["health_signals"] = _recent_signals(conn, deployment_id)
+    return result
+
+
+def _recent_signals(conn, deployment_id, limit=50):
+    rows = conn.execute(
+        "SELECT signal_id, sampled_at, signal_type, status, detail FROM deploy_health_signal "
+        "WHERE deployment_id = ? ORDER BY rowid DESC LIMIT ?",
+        (deployment_id, limit),
+    ).fetchall()
+    return [
+        {"signal_id": r[0], "sampled_at": r[1], "signal_type": r[2], "status": r[3], "detail": r[4]}
+        for r in rows
+    ]
@@ -948,6 +948,73 @@ class ArtifactRouteTest(unittest.TestCase):
         self.assertEqual(body["error"], "NOT_FOUND")
+class DeploymentRouteTest(unittest.TestCase):
+    """U-043: c_deployment_list/c_deployment_current/c_deployment_get for
+    v_deployment_status."""
+
+    def setUp(self):
+        self.server = backend_server.create_server(db_path=":memory:")
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        conn = self.server.conn
+        conn.execute(
+            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
+            "previous_instance_dir) VALUES ('dep-1', 'c1', 'c0', 'finalized', '/prev')"
+        )
+        conn.execute(
+            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
+            "previous_instance_dir) VALUES ('dep-2', 'c2', 'c1', 'monitoring', '/prev2')"
+        )
+        conn.execute(
+            "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
+            "VALUES ('s1', 'dep-2', CURRENT_TIMESTAMP, 'process_alive', 'ok')"
+        )
+        conn.commit()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+
+    def _get(self, path):
+        host, port = self.server.server_address
+        conn = http.client.HTTPConnection(host, port, timeout=5)
+        conn.request("GET", path)
+        resp = conn.getresponse()
+        body = json.loads(resp.read())
+        conn.close()
+        return resp.status, body
+
+    def test_list_returns_all_newest_first(self):
+        status, body = self._get("/api/deployments")
+        self.assertEqual(status, 200)
+        self.assertEqual([d["deployment_id"] for d in body], ["dep-2", "dep-1"])
+
+    def test_current_returns_the_active_one(self):
+        status, body = self._get("/api/deployments/current")
+        self.assertEqual(status, 200)
+        self.assertEqual(body["deployment_id"], "dep-2")
+
+    def test_current_is_null_when_nothing_active(self):
+        with self.server.conn_lock:
+            self.server.conn.execute("UPDATE deployment SET status = 'finalized' WHERE deployment_id = 'dep-2'")
+            self.server.conn.commit()
+        status, body = self._get("/api/deployments/current")
+        self.assertEqual(status, 200)
+        self.assertIsNone(body)
+
+    def test_get_includes_health_signals(self):
+        status, body = self._get("/api/deployments/dep-2")
+        self.assertEqual(status, 200)
+        self.assertEqual(len(body["health_signals"]), 1)
+        self.assertEqual(body["health_signals"][0]["signal_id"], "s1")
+
+    def test_get_nonexistent_returns_404(self):
+        status, body = self._get("/api/deployments/no-such-deployment")
+        self.assertEqual(status, 404)
+        self.assertEqual(body["error"], "NOT_FOUND")
+
+
 class SkillRouteTest(unittest.TestCase):
     """U-029: c_skill_diff (GET /api/skills/diff) + c_skill_rollback
     (POST /api/skills/:skill_id/rollback) + list/get for v_skill_library /
+"""U-043 tests: deployment_query.py. All real (no mocking needed -- sqlite
+is fast/free/local).
+
+Run: python3 -m unittest src.test_deployment_query -v
+"""
+
+import unittest
+
+from src import deployment_query, schema
+
+
+class DeploymentQueryTestBase(unittest.TestCase):
+    def setUp(self):
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+
+    def tearDown(self):
+        self.conn.close()
+
+    def _insert(self, deployment_id, status, **kwargs):
+        cols = ["deployment_id", "git_commit_hash", "previous_commit_hash", "status", "previous_instance_dir"]
+        vals = [deployment_id, kwargs.get("git_commit_hash", "c1"), kwargs.get("previous_commit_hash", "c0"),
+                status, kwargs.get("previous_instance_dir", "/prev")]
+        for extra_col in ("candidate_port", "candidate_instance_dir", "started_at", "cutover_at",
+                          "monitoring_closed_at", "rollback_at", "rollback_reason"):
+            if extra_col in kwargs:
+                cols.append(extra_col)
+                vals.append(kwargs[extra_col])
+        self.conn.execute(
+            "INSERT INTO deployment ({}) VALUES ({})".format(", ".join(cols), ", ".join("?" * len(cols))), vals
+        )
+        self.conn.commit()
+
+
+class ListDeploymentsTest(DeploymentQueryTestBase):
+    def test_newest_first(self):
+        self._insert("d1", "finalized")
+        self._insert("d2", "finalized")
+        result = deployment_query.list_deployments(self.conn)
+        self.assertEqual([d["deployment_id"] for d in result], ["d2", "d1"])
+
+    def test_respects_limit(self):
+        for i in range(5):
+            self._insert("d{}".format(i), "finalized")
+        result = deployment_query.list_deployments(self.conn, limit=2)
+        self.assertEqual(len(result), 2)
+
+    def test_is_active_flag(self):
+        self._insert("d1", "monitoring")
+        self._insert("d2", "finalized")
+        by_id = {d["deployment_id"]: d for d in deployment_query.list_deployments(self.conn)}
+        self.assertTrue(by_id["d1"]["is_active"])
+        self.assertFalse(by_id["d2"]["is_active"])
+
+
+class GetCurrentDeploymentTest(DeploymentQueryTestBase):
+    def test_none_when_nothing_active(self):
+        self._insert("d1", "finalized")
+        self.assertIsNone(deployment_query.get_current_deployment(self.conn))
+
+    def test_returns_the_one_active_deployment(self):
+        self._insert("d1", "finalized")
+        self._insert("d2", "health_check_running")
+        result = deployment_query.get_current_deployment(self.conn)
+        self.assertEqual(result["deployment_id"], "d2")
+        self.assertTrue(result["is_active"])
+
+    def test_blocked_pending_human_is_not_active(self):
+        # 01-M 1.15's ux_deployment_active constraint treats blocked_pending_human
+        # as terminal too -- it's the one status that needs a human, not the
+        # scheduler, to move it forward again.
+        self._insert("d1", "blocked_pending_human")
+        self.assertIsNone(deployment_query.get_current_deployment(self.conn))
+
+
+class GetDeploymentTest(DeploymentQueryTestBase):
+    def test_not_found_raises(self):
+        with self.assertRaises(deployment_query.DeploymentNotFoundError):
+            deployment_query.get_deployment(self.conn, "nonexistent")
+
+    def test_includes_recent_health_signals_newest_first(self):
+        self._insert("d1", "monitoring")
+        self.conn.execute(
+            "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
+            "VALUES ('s1', 'd1', '2020-01-01 00:00:00', 'process_alive', 'ok')"
+        )
+        self.conn.execute(
+            "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
+            "VALUES ('s2', 'd1', '2020-01-01 00:05:00', 'http_health', 'ok')"
+        )
+        self.conn.commit()
+        result = deployment_query.get_deployment(self.conn, "d1")
+        self.assertEqual([s["signal_id"] for s in result["health_signals"]], ["s2", "s1"])
+
+    def test_no_signals_yet_gives_empty_list(self):
+        self._insert("d1", "staged")
+        result = deployment_query.get_deployment(self.conn, "d1")
+        self.assertEqual(result["health_signals"], [])
+
+
+if __name__ == "__main__":
+    unittest.main()