get_drift_signature

Instructions

Implementation Reference

• src/mk_spec_master/tools/history.py:153-241 (handler)

  The handler function get_drift_signature_tool that implements the tool logic. It reads snapshot history, counts chronic appearances of spec_ids across drifted/unknown/quality buckets, and returns a drift signature report with markdown output.

  def get_drift_signature_tool(arguments: dict) -> dict[str, Any]:
      """Scan history for chronic problems: same spec_id repeatedly showing
      up in drifted / unknown / low-quality buckets.
  
      Threshold defaults: appears in >= 3 of the last 5 snapshots → chronic.
  
      Args:
          window: int, default 5 — how many recent snapshots to scan.
          threshold: int, default 3 — minimum recurrence to flag as chronic.
      """
      window = int(arguments.get("window", 5))
      threshold = int(arguments.get("threshold", 3))
  
      snapshots = _read_snapshots(limit=window)
      if len(snapshots) < threshold:
          return {
              "ready": False,
              "snapshots_available": len(snapshots),
              "snapshots_needed": threshold,
              "chronic": [],
              "markdown": (
                  "# Drift signature\n\n"
                  f"_Need at least {threshold} snapshots; have {len(snapshots)}. "
                  f"Run `get_optimization_plan` over time to build history._"
              ),
          }
  
      drift_counts: dict[str, int] = {}
      unknown_counts: dict[str, int] = {}
      quality_counts: dict[str, int] = {}
  
      for snap in snapshots:
          for d in snap.get("drifted", []) or []:
              sid = d.get("spec_id")
              if sid:
                  drift_counts[sid] = drift_counts.get(sid, 0) + 1
          for u in snap.get("unknown", []) or []:
              sid = u.get("spec_id")
              if sid:
                  unknown_counts[sid] = unknown_counts.get(sid, 0) + 1
          for q in snap.get("quality", []) or []:
              sid = q.get("spec_id")
              if sid:
                  quality_counts[sid] = quality_counts.get(sid, 0) + 1
  
      chronic = []
      for sid, count in drift_counts.items():
          if count >= threshold:
              chronic.append({"spec_id": sid, "kind": "unstable", "appearances": count, "window": len(snapshots),
                              "reason": "Spec keeps changing after being linked — PM may be iterating without resetting downstream tests."})
      for sid, count in quality_counts.items():
          if count >= threshold:
              chronic.append({"spec_id": sid, "kind": "chronic_low_quality", "appearances": count, "window": len(snapshots),
                              "reason": "Spec is repeatedly flagged for vague language / implementation-leak / unclear roles."})
      for sid, count in unknown_counts.items():
          if count >= threshold:
              chronic.append({"spec_id": sid, "kind": "chronic_unhashed", "appearances": count, "window": len(snapshots),
                              "reason": "Linked tests never recorded an ac_hash — re-link with parse_spec._meta.ac_hash to enable drift detection."})
  
      chronic.sort(key=lambda r: (-r["appearances"], r["spec_id"]))
  
      md = ["# Drift signature", ""]
      md.append(f"- Window: last {len(snapshots)} snapshots")
      md.append(f"- Threshold: ≥ {threshold} appearances = chronic")
      md.append(f"- Chronic specs flagged: {len(chronic)}")
      md.append("")
      if not chronic:
          md.append("🟢 No chronic patterns detected. Keep an eye on the next few snapshots.")
      else:
          kind_label = {
              "unstable": "🔴 Unstable (repeatedly drifting)",
              "chronic_low_quality": "🟡 Chronically low quality",
              "chronic_unhashed": "⚪ Chronically without ac_hash",
          }
          last_kind = None
          for c in chronic:
              if c["kind"] != last_kind:
                  md.append("")
                  md.append(f"## {kind_label.get(c['kind'], c['kind'])}")
                  last_kind = c["kind"]
              md.append(f"- `{c['spec_id']}` — appeared {c['appearances']}/{c['window']} snapshots · {c['reason']}")
  
      return {
          "ready": True,
          "snapshots_scanned": len(snapshots),
          "threshold": threshold,
          "chronic": chronic,
          "markdown": "\n".join(md),
      }

• src/mk_spec_master/server.py:44-63 (registration)

  The dispatch table _DISPATCH that maps the string 'get_drift_signature' to history_tools.get_drift_signature_tool for tool routing.

  _DISPATCH: dict[str, Callable[[dict], dict]] = {
      "get_spec_source_info": _meta_info,
      "list_specs": specs_tools.list_specs_tool,
      "fetch_spec": specs_tools.fetch_spec_tool,
      "parse_spec": specs_tools.parse_spec_tool,
      "extract_scenarios": scenarios_tools.extract_scenarios_tool,
      "generate_test_plan": scenarios_tools.generate_test_plan_tool,
      "link_test_to_spec": coverage_tools.link_test_to_spec_tool,
      "get_coverage_matrix": coverage_tools.get_coverage_matrix_tool,
      "get_drift_report": coverage_tools.get_drift_report_tool,
      "analyze_spec_quality": quality_tools.analyze_spec_quality_tool,
      "propose_spec_improvements": quality_tools.propose_spec_improvements_tool,
      "auto_link_tests": auto_link_tools.auto_link_tests_tool,
      "get_optimization_plan": optimization_tools.get_optimization_plan_tool,
      "init_spec_knowledge": spec_knowledge_tools.init_spec_knowledge_tool,
      "get_spec_context": spec_knowledge_tools.get_spec_context_tool,
      "get_spec_history": history_tools.get_spec_history_tool,
      "get_drift_signature": history_tools.get_drift_signature_tool,
      "get_telemetry": telemetry_tools.get_telemetry_tool,
  }

• src/mk_spec_master/server.py:433-454 (registration)

  The Tool registration via @app.list_tools() that defines the name, description, and inputSchema (window, threshold) for the 'get_drift_signature' MCP tool.

  Tool(
      name="get_drift_signature",
      description=(
          "Scan the recent snapshot history for chronic problems: "
          "same spec_id repeatedly appearing in drifted / unknown / "
          "low-quality buckets. Specs flagged as 'unstable' (drifts "
          "every cycle), 'chronic_low_quality' (vague every cycle), "
          "or 'chronic_unhashed' (never gets a hash recorded). "
          "Use when a user asks 'which specs keep causing trouble' / "
          "'what's the long-running pain'. "
          "Args: window (snapshots to scan, default 5), threshold "
          "(min recurrence to flag, default 3). "
          "Returns {ready, snapshots_scanned, chronic[], markdown}."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "window": {"type": "integer", "default": 5},
              "threshold": {"type": "integer", "default": 3},
          },
      },
  ),

• src/mk_spec_master/tools/history.py:43-57 (helper)

  The helper _read_snapshots that reads snapshot JSON files from HISTORY_DIR, used by get_drift_signature_tool to load history data.

  def _read_snapshots(limit: int | None = None) -> list[dict]:
      """Return snapshots in chronological order (oldest first). Reads the
      directory listing; tolerates missing dir + bad files."""
      if not config.HISTORY_DIR.exists():
          return []
      files = sorted(config.HISTORY_DIR.glob("*.json"))
      if limit:
          files = files[-limit:]
      out: list[dict] = []
      for f in files:
          try:
              out.append(json.loads(f.read_text(encoding="utf-8")))
          except (OSError, json.JSONDecodeError):
              continue
      return out