Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 66 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(control-plane): persist allocator dependency edges as durable state (Closes #784)

Browse Source 
Umbrella #628 scope item 6 requires dependencies to be durable structured
state carrying source, target, type, blocking condition, completion
condition, current state, and evidence. Nothing stored any of that.

Dependency knowledge existed only as a per-run computation:
allocator_dependencies re-parsed the Depends: declaration out of every
issue body on every allocation, _allocator_candidates_from_gitea resolved
each reference against live issue state, and the result collapsed into two
in-memory WorkCandidate fields that classify_skip consumed and discarded.
Three consequences followed. Nothing could answer "what is waiting on #N"
without re-listing every open issue and re-parsing every body, so the
reverse edge automatic resumption needs did not exist in any form. Only
issue-blocked-by-issue was expressible, leaving the other six #628
relationships with nowhere to live. And no observation was recorded, so a
transient lookup failure and a real block were indistinguishable after the
fact.

Add the store:

- dependency_edges table under schema v4. Creating the table is itself the
  v3 to v4 migration: additive, idempotent, and it never touches the
  existing tables. Uniqueness is (scope, source, target, edge_type), so
  re-observation updates one row rather than appending duplicates.
- dependency_graph.py owns the vocabulary: the seven #628 relationship
  types, the three states, and fail-closed normalization for both plus
  endpoint kinds. An unrecognized value writes nothing rather than landing
  as unqueryable free text. Evidence is sanitized before storage, so no
  credential or endpoint URL can be persisted or read back.
- upsert_dependency_edge, list_dependency_edges, and
  record_dependency_edge_observation on ControlPlaneDB. Filtering by target
  makes reverse lookup a single query. State transitions append to the
  existing events table rather than a parallel audit table.
- The allocator persists what it already resolved. States map one-to-one
  from the resolver's met/unmet/unavailable partitions, so nothing is
  re-classified and unavailable evidence is never recorded as met.
- gitea_list_dependency_edges exposes stored edges read-only, gated on
  gitea.read, and is added to the documented inventory the #781 drift guard
  checks.

Selection is deliberately untouched: classify_skip still consumes the
in-memory dependency_unmet field. The write is best-effort and reports
failures through reasons, so a broken or absent store leaves allocation
behaving exactly as it did before — proven by allocating the same candidate
set through a store whose writes all raise and comparing the selection,
skip set, and candidate count.

Automatic blocking and resumption (#628 item 7), non-issue edge creation,
and defect auto-linking are later slices; this one only makes the graph
durable and queryable.

Tests: 31 new cases covering fresh-schema creation, a real v3-to-v4
migration with row retention, idempotent re-migration, enum rejection,
upsert idempotence, forward and reverse lookup, scope isolation, transition
events, redaction at rest, live allocation-run ingestion, write-failure
tolerance, and the tool's permission gate.

Verification: 4126 passed in the branch worktree. The 11 failures in
test_commit_payloads, test_issue_702_review_findings_f1_f6, test_mcp_server,
test_post_merge_moot_lease, and test_reconciler_supersession_close reproduce
identically on a clean detached checkout of master at 300e8acd, so they are
pre-existing and proven by baseline run, not introduced here.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-21 18:07:21 -04:00
co-authored by Claude Opus 4.8
parent 300e8acd13
commit 0589ec8069
6 changed files with 1475 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
control_plane_db.py
+326 -1
View File
@@ -29,7 +29,9 @@ from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Iterator, Sequence
SCHEMA_VERSION = 3
import dependency_graph
SCHEMA_VERSION = 4
# Assignable work kinds only — raw monitoring incidents are never work items.
WORK_KINDS = frozenset({"issue", "pr"})
@@ -147,7 +149,41 @@ CREATE TABLE IF NOT EXISTS incident_links (
UNIQUE (provider, provider_base_url, provider_org, provider_project, provider_issue_id)
);
-- Durable dependency graph (#784, umbrella #628 scope item 6). Dependencies
-- were previously re-parsed per allocation run and discarded; each row here is
-- one relationship with its conditions, current state, and evidence. Creating
-- the table is itself the v3→v4 migration: additive, idempotent, and it never
-- touches the pre-existing tables.
CREATE TABLE IF NOT EXISTS dependency_edges (
edge_id TEXT PRIMARY KEY,
remote TEXT NOT NULL,
org TEXT NOT NULL,
repo TEXT NOT NULL,
source_kind TEXT NOT NULL CHECK (source_kind IN ('issue', 'pr')),
source_number INTEGER NOT NULL,
target_kind TEXT NOT NULL CHECK (target_kind IN ('issue', 'pr')),
target_number INTEGER NOT NULL,
edge_type TEXT NOT NULL,
blocking_condition TEXT NOT NULL DEFAULT '',
completion_condition TEXT NOT NULL DEFAULT '',
state TEXT NOT NULL,
evidence TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
last_observed_at TEXT NOT NULL,
UNIQUE (
remote, org, repo, source_kind, source_number,
target_kind, target_number, edge_type
)
);
CREATE INDEX IF NOT EXISTS idx_leases_work_status ON leases(work_item_id, status);
-- Reverse lookup ("what waits on this target") is the query automatic
-- resumption needs, so it gets its own index alongside the forward one.
CREATE INDEX IF NOT EXISTS idx_dependency_edges_source
ON dependency_edges(remote, org, repo, source_kind, source_number);
CREATE INDEX IF NOT EXISTS idx_dependency_edges_target
ON dependency_edges(remote, org, repo, target_kind, target_number);
CREATE INDEX IF NOT EXISTS idx_assignments_session ON assignments(session_id, status);
CREATE INDEX IF NOT EXISTS idx_incident_gitea ON incident_links(gitea_org, gitea_repo, gitea_issue_number);
"""
@@ -1848,3 +1884,292 @@ class ControlPlaneDB:
f"transferred lease ownership from {owner} to {adopter_session_id}"
],
}
# --- Dependency graph (#784, umbrella #628 scope item 6) ----------------
@staticmethod
def _dependency_edge_row(row: sqlite3.Row | None) -> dict[str, Any] | None:
"""Return a stored edge as a plain dict with evidence decoded."""
if row is None:
return None
edge = dict(row)
raw = edge.get("evidence")
try:
edge["evidence"] = json.loads(raw) if raw else {}
except (TypeError, ValueError):
# A row written by an older/foreign writer must not break reads.
edge["evidence"] = {"unparsed": str(raw)}
return edge
def upsert_dependency_edge(
self,
*,
remote: str,
org: str,
repo: str,
source_kind: str,
source_number: int,
target_kind: str,
target_number: int,
edge_type: str,
state: str,
blocking_condition: str | None = None,
completion_condition: str | None = None,
evidence: Any = None,
) -> dict[str, Any]:
"""Insert or refresh one dependency edge, keyed by its relationship.
Uniqueness is (scope, source, target, edge_type), so re-observing the
same relationship updates one row instead of appending history — the
edge is current state, and transitions are recorded as ``events``.
Edge type, state, and both endpoint kinds are validated fail-closed;
an unknown value writes nothing. Evidence is sanitized before storage.
"""
edge_type_norm = dependency_graph.normalize_edge_type(edge_type)
state_norm = dependency_graph.normalize_edge_state(state)
source_kind_norm = dependency_graph.normalize_work_kind(source_kind)
target_kind_norm = dependency_graph.normalize_work_kind(target_kind)
source_no = int(source_number)
target_no = int(target_number)
if blocking_condition is None or completion_condition is None:
defaults = dependency_graph.default_conditions(edge_type_norm)
blocking_condition = (
defaults[0] if blocking_condition is None else blocking_condition
)
completion_condition = (
defaults[1] if completion_condition is None else completion_condition
)
evidence_json = json.dumps(
dependency_graph.sanitize_evidence(evidence if evidence is not None else {})
)
now_s = _ts()
with self._tx() as conn:
existing = conn.execute(
"""
SELECT * FROM dependency_edges
WHERE remote = ? AND org = ? AND repo = ?
AND source_kind = ? AND source_number = ?
AND target_kind = ? AND target_number = ? AND edge_type = ?
""",
(
remote,
org,
repo,
source_kind_norm,
source_no,
target_kind_norm,
target_no,
edge_type_norm,
),
).fetchone()
if existing is None:
edge_id = uuid.uuid4().hex
conn.execute(
"""
INSERT INTO dependency_edges(
edge_id, remote, org, repo,
source_kind, source_number, target_kind, target_number,
edge_type, blocking_condition, completion_condition,
state, evidence, created_at, updated_at, last_observed_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
edge_id,
remote,
org,
repo,
source_kind_norm,
source_no,
target_kind_norm,
target_no,
edge_type_norm,
blocking_condition,
completion_condition,
state_norm,
evidence_json,
now_s,
now_s,
now_s,
),
)
else:
edge_id = str(existing["edge_id"])
conn.execute(
"""
UPDATE dependency_edges
SET blocking_condition = ?, completion_condition = ?,
state = ?, evidence = ?, updated_at = ?,
last_observed_at = ?
WHERE edge_id = ?
""",
(
blocking_condition,
completion_condition,
state_norm,
evidence_json,
now_s,
now_s,
edge_id,
),
)
prior_state = str(existing["state"])
if prior_state != state_norm:
self._record_edge_transition_conn(
conn,
edge_id=edge_id,
prior_state=prior_state,
new_state=state_norm,
detail="observed during upsert",
now_s=now_s,
)
row = conn.execute(
"SELECT * FROM dependency_edges WHERE edge_id = ?", (edge_id,)
).fetchone()
return self._dependency_edge_row(row) or {}
@staticmethod
def _record_edge_transition_conn(
conn: sqlite3.Connection,
*,
edge_id: str,
prior_state: str,
new_state: str,
detail: str,
now_s: str,
) -> None:
"""Append a state transition to the shared ``events`` audit table.
``work_item_id`` stays NULL: an edge endpoint is a Gitea issue/PR that
may never have been assigned, so it has no work_items row to reference.
"""
message = (
f"dependency edge {edge_id} state {prior_state} -> {new_state}"
f" ({detail})"
)
conn.execute(
"""
INSERT INTO events(work_item_id, event_type, message, created_at)
VALUES (NULL, 'dependency_edge_state_change', ?, ?)
""",
(message, now_s),
)
def list_dependency_edges(
self,
*,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
source_kind: str | None = None,
source_number: int | None = None,
target_kind: str | None = None,
target_number: int | None = None,
edge_type: str | None = None,
state: str | None = None,
limit: int = 500,
) -> list[dict[str, Any]]:
"""Return stored edges, filtered.
Filtering by *target* answers "what is waiting on this work unit",
which is the query automatic resumption needs and which body-text
parsing could never serve.
"""
clauses: list[str] = []
params: list[Any] = []
if remote:
clauses.append("remote = ?")
params.append(remote)
if org:
clauses.append("org = ?")
params.append(org)
if repo:
clauses.append("repo = ?")
params.append(repo)
if source_kind:
clauses.append("source_kind = ?")
params.append(dependency_graph.normalize_work_kind(source_kind))
if source_number is not None:
clauses.append("source_number = ?")
params.append(int(source_number))
if target_kind:
clauses.append("target_kind = ?")
params.append(dependency_graph.normalize_work_kind(target_kind))
if target_number is not None:
clauses.append("target_number = ?")
params.append(int(target_number))
if edge_type:
clauses.append("edge_type = ?")
params.append(dependency_graph.normalize_edge_type(edge_type))
if state:
clauses.append("state = ?")
params.append(dependency_graph.normalize_edge_state(state))
sql = "SELECT * FROM dependency_edges"
if clauses:
sql += " WHERE " + " AND ".join(clauses)
sql += " ORDER BY source_number ASC, target_number ASC, edge_type ASC LIMIT ?"
params.append(int(limit))
with self._tx(immediate=False) as conn:
rows = conn.execute(sql, params).fetchall()
return [edge for edge in (self._dependency_edge_row(r) for r in rows) if edge]
def record_dependency_edge_observation(
self,
edge_id: str,
*,
state: str,
evidence: Any = None,
detail: str = "observation recorded",
) -> dict[str, Any]:
"""Update an existing edge's state and evidence, auditing the change.
A transition writes an ``events`` row carrying both the prior and the
new state, so a later blocked/resume decision can be reconstructed from
durable state rather than from a recomputed reason string.
"""
state_norm = dependency_graph.normalize_edge_state(state)
now_s = _ts()
with self._tx() as conn:
existing = conn.execute(
"SELECT * FROM dependency_edges WHERE edge_id = ?", (edge_id,)
).fetchone()
if existing is None:
raise ControlPlaneError(
f"dependency edge '{edge_id}' does not exist (fail closed)"
)
prior_state = str(existing["state"])
if evidence is None:
evidence_json = str(existing["evidence"] or "{}")
else:
evidence_json = json.dumps(
dependency_graph.sanitize_evidence(evidence)
)
conn.execute(
"""
UPDATE dependency_edges
SET state = ?, evidence = ?, updated_at = ?, last_observed_at = ?
WHERE edge_id = ?
""",
(state_norm, evidence_json, now_s, now_s, edge_id),
)
if prior_state != state_norm:
self._record_edge_transition_conn(
conn,
edge_id=edge_id,
prior_state=prior_state,
new_state=state_norm,
detail=detail,
now_s=now_s,
)
row = conn.execute(
"SELECT * FROM dependency_edges WHERE edge_id = ?", (edge_id,)
).fetchone()
edge = self._dependency_edge_row(row) or {}
edge["prior_state"] = prior_state
edge["state_changed"] = prior_state != state_norm
return edge