Coverage for src/keel/workcreation.py: 100%
88 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Deterministic policy for signal-driven work creation."""
3from __future__ import annotations
5import re
6from dataclasses import dataclass
7from typing import Any
9SCHEMA_VERSION = "keel.work-creation.v1"
10DEFAULT_MIN_OCCURRENCES = 2
11DEFAULT_MIN_CONFIDENCE = 0.6
12DEFAULT_MAX_CREATIONS = 5
13DEFAULT_NEAR_TEXT_SIMILARITY = 0.6
15DECISIONS = (
16 "create",
17 "suppress-transient",
18 "suppress-duplicate",
19 "limit-reached",
20)
22_TOKEN_RE = re.compile(r"[a-z0-9]+")
25@dataclass(frozen=True)
26class WorkDecision:
27 """One deterministic work-creation decision."""
29 candidate_id: str
30 decision: str
31 reason: str
32 title: str
33 duplicate_of: int | None = None
35 def as_dict(self) -> dict[str, Any]:
36 result: dict[str, Any] = {
37 "candidate_id": self.candidate_id,
38 "decision": self.decision,
39 "reason": self.reason,
40 "title": self.title,
41 "creates_issue": self.decision == "create",
42 }
43 if self.duplicate_of is not None:
44 result["duplicate_of"] = self.duplicate_of
45 return result
48def contract_as_dict(*, near_text_similarity: float | None = None) -> dict[str, Any]:
49 """Return the shared work-creation policy contract.
51 ``near_text_similarity`` is the project's resolved
52 ``policy_pack.scan.near_text_similarity``. It has to be passed in rather than
53 defaulted here, because the caller embeds this dict *beside* a `dedupe` block that
54 already honours the knob — hardcoding the default shipped two different thresholds
55 under near-identical keys in one contract, agreeing only as long as the project set
56 the knob to exactly the built-in default (#633).
57 """
58 return {
59 "schema_version": SCHEMA_VERSION,
60 "consumer_neutral": True,
61 "deterministic": True,
62 "stdlib_only": True,
63 "source": "signal-driven commands",
64 "decisions": list(DECISIONS),
65 "transient_filter": {
66 "default_min_occurrences": DEFAULT_MIN_OCCURRENCES,
67 "default_min_confidence": DEFAULT_MIN_CONFIDENCE,
68 "transient_outcome": "suppress-transient",
69 },
70 "dedupe": {
71 "against": "open work",
72 "keys": ["dedupe_key", "normalized_title", "near_text"],
73 "near_text_similarity": _confidence(
74 near_text_similarity, DEFAULT_NEAR_TEXT_SIMILARITY
75 ),
76 "duplicate_outcome": "suppress-duplicate",
77 },
78 "cycle_limit": {
79 "default_max_creations": DEFAULT_MAX_CREATIONS,
80 "limit_outcome": "limit-reached",
81 },
82 "consumers": [
83 "regression",
84 "review-all-day",
85 "coverage",
86 "deps-audit",
87 "flake-audit",
88 ],
89 }
92def evaluate_candidates(
93 candidates: list[dict[str, Any]] | tuple[dict[str, Any], ...],
94 existing_work: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (),
95 *,
96 min_occurrences: int = DEFAULT_MIN_OCCURRENCES,
97 min_confidence: float = DEFAULT_MIN_CONFIDENCE,
98 max_creations: int = DEFAULT_MAX_CREATIONS,
99 near_text_similarity: float = DEFAULT_NEAR_TEXT_SIMILARITY,
100) -> dict[str, Any]:
101 """Evaluate candidate signals against transient, dedupe, and cycle-limit policy."""
102 policy = {
103 "min_occurrences": _positive_int(min_occurrences, DEFAULT_MIN_OCCURRENCES),
104 "min_confidence": _confidence(min_confidence, DEFAULT_MIN_CONFIDENCE),
105 "max_creations": _positive_int(max_creations, DEFAULT_MAX_CREATIONS),
106 "near_text_similarity": _confidence(
107 near_text_similarity,
108 DEFAULT_NEAR_TEXT_SIMILARITY,
109 ),
110 }
111 normalized_existing = [
112 _normalize_existing(item)
113 for item in existing_work
114 if isinstance(item, dict) and _is_open(item)
115 ]
116 created = 0
117 decisions: list[WorkDecision] = []
118 for index, raw in enumerate(candidates, start=1):
119 if not isinstance(raw, dict):
120 continue
121 candidate = _normalize_candidate(raw, index)
122 duplicate = _find_duplicate(candidate, normalized_existing, policy["near_text_similarity"])
123 if _is_transient(candidate, policy):
124 decisions.append(WorkDecision(
125 candidate["id"],
126 "suppress-transient",
127 "transient-signal",
128 candidate["title"],
129 ))
130 elif duplicate is not None:
131 decisions.append(WorkDecision(
132 candidate["id"],
133 "suppress-duplicate",
134 "open-work-duplicate",
135 candidate["title"],
136 duplicate_of=duplicate["number"],
137 ))
138 elif created >= policy["max_creations"]:
139 decisions.append(WorkDecision(
140 candidate["id"],
141 "limit-reached",
142 "per-cycle-limit-reached",
143 candidate["title"],
144 ))
145 else:
146 created += 1
147 decisions.append(WorkDecision(
148 candidate["id"],
149 "create",
150 "eligible",
151 candidate["title"],
152 ))
153 normalized_existing.append(_created_as_existing(candidate, created))
154 decision_dicts = [decision.as_dict() for decision in decisions]
155 return {
156 "schema_version": SCHEMA_VERSION,
157 "status": "pass",
158 "policy": policy,
159 "summary": {
160 "candidates": len([item for item in candidates if isinstance(item, dict)]),
161 "create": _count(decision_dicts, "create"),
162 "suppress_transient": _count(decision_dicts, "suppress-transient"),
163 "suppress_duplicate": _count(decision_dicts, "suppress-duplicate"),
164 "limit_reached": _count(decision_dicts, "limit-reached"),
165 },
166 "decisions": decision_dicts,
167 }
170def _normalize_candidate(raw: dict[str, Any], index: int) -> dict[str, Any]:
171 title = _string(raw.get("title")) or f"candidate-{index}"
172 body = _string(raw.get("body"))
173 return {
174 "id": _string(raw.get("id")) or f"candidate-{index}",
175 "title": title,
176 "body": body,
177 "dedupe_key": _string(raw.get("dedupe_key")),
178 "occurrences": _positive_int(raw.get("occurrences"), 1),
179 "confidence": _confidence(raw.get("confidence"), 1.0),
180 "normalized_title": _normalize_text(title),
181 "tokens": _tokens(f"{title} {body}"),
182 }
185def _normalize_existing(raw: dict[str, Any]) -> dict[str, Any]:
186 title = _string(raw.get("title"))
187 body = _string(raw.get("body"))
188 return {
189 "number": raw.get("number") if isinstance(raw.get("number"), int) else None,
190 "title": title,
191 "body": body,
192 "dedupe_key": _string(raw.get("dedupe_key")),
193 "normalized_title": _normalize_text(title),
194 "tokens": _tokens(f"{title} {body}"),
195 }
198def _created_as_existing(candidate: dict[str, Any], created_index: int) -> dict[str, Any]:
199 return {
200 "number": -created_index,
201 "title": candidate["title"],
202 "body": candidate["body"],
203 "dedupe_key": candidate["dedupe_key"],
204 "normalized_title": candidate["normalized_title"],
205 "tokens": candidate["tokens"],
206 }
209def _is_transient(candidate: dict[str, Any], policy: dict[str, Any]) -> bool:
210 return (
211 candidate["occurrences"] < policy["min_occurrences"]
212 or candidate["confidence"] < policy["min_confidence"]
213 )
216def _find_duplicate(
217 candidate: dict[str, Any],
218 existing: list[dict[str, Any]],
219 threshold: float,
220) -> dict[str, Any] | None:
221 for item in existing:
222 if _same_key(candidate, item) or _same_title(candidate, item):
223 return item
224 if _jaccard(candidate["tokens"], item["tokens"]) >= threshold:
225 return item
226 return None
229def _same_key(candidate: dict[str, Any], existing: dict[str, Any]) -> bool:
230 return bool(candidate["dedupe_key"] and candidate["dedupe_key"] == existing["dedupe_key"])
233def _same_title(candidate: dict[str, Any], existing: dict[str, Any]) -> bool:
234 return bool(
235 candidate["normalized_title"]
236 and candidate["normalized_title"] == existing["normalized_title"]
237 )
240def _is_open(raw: dict[str, Any]) -> bool:
241 state = _string(raw.get("state")).lower()
242 # Missing state is treated as open so incomplete GitHub/search fixtures
243 # suppress duplicates conservatively instead of creating duplicate work.
244 return state in {"", "open"}
247def _tokens(text: str) -> set[str]:
248 return set(_TOKEN_RE.findall(_normalize_text(text)))
251def _jaccard(left: set[str], right: set[str]) -> float:
252 if not left or not right:
253 return 0.0
254 return len(left & right) / len(left | right)
257def _normalize_text(text: str) -> str:
258 return " ".join(_TOKEN_RE.findall(text.lower()))
261def _string(value: Any) -> str:
262 return value.strip() if isinstance(value, str) and value.strip() else ""
265def _positive_int(value: Any, default: int) -> int:
266 return value if isinstance(value, int) and value > 0 else default
269def _confidence(value: Any, default: float) -> float:
270 return value if isinstance(value, int | float) and 0 <= value <= 1 else default
273def _count(decisions: list[dict[str, Any]], decision: str) -> int:
274 return sum(item["decision"] == decision for item in decisions)