self_hosting_e2e.py
6.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
"""U-042: self-hosting closed-loop acceptance -- wires every PH5 module
built so far (U-034~U-041) plus the Graph Engine's own scheduler into the
literal 9-step scenario `self-development-feasibility.md` walks through:
a real self-target Goal, Planner->Coder->Validator (real `claude` calls via
`scheduler.run_goal_to_completion`, which already forces worktree isolation
and records the `core_engine` audit trail for self-target tasks per U-035),
merge, stage a candidate, health-check it, cut it over, monitor it, and
land on EITHER `finalized` (clean monitoring window) OR a rollback (anomaly
detected) -- "走完整...两条路径中至少一条", not a scripted single outcome.
This module is the acceptance scenario's own orchestration glue, not a
claim about what Deploy Supervisor's real production `poll_fn` (U-036's
still-unfilled hook) should look like long-term -- a continuously-running
daemon would discover self-target merges to stage on its own schedule
rather than drive one linear call end to end. Building that discovery loop
is future work, not this Unit's job; this function exists so the acceptance
scenario has one real, callable entry point to run (or NOT run -- see
test_self_hosting_e2e.py's gating) against.
"""
import time
from supervisor import deploy_cutover, deploy_health_check, deploy_monitor, deploy_rollback, deploy_stage
from src import paths, schema, scheduler
class SelfHostingLoopError(Exception):
"""Raised when a precondition for even attempting the scenario isn't
met (e.g. target_folder_path isn't genuinely self-target) -- distinct
from a real outcome like 'rolled_back', which is a valid, successful
exercise of the safety mechanism, not an error."""
def run_self_hosting_closed_loop(
db_path, goal_text, target_folder_path=None, prod_db_path=None,
poll_interval=5.0, monitor_timeout_cycles=None, sleep=time.sleep, now=time.time,
):
"""Returns {"stage", "outcome", ...}. `stage` names how far the scenario
got (`task_lifecycle` / `classification` / `health_check` / `cutover` /
`monitor`); `outcome` is the concrete result at that stage. A task that
doesn't reach 'done', or doesn't get classified core_engine, is a valid
early-return (not every self-target task is a core-engine change -- see
U-035), not a scenario failure.
"""
target_folder_path = target_folder_path or paths.repo_root()
prod_db_path = prod_db_path or paths.db_path()
if not paths.is_self_target(target_folder_path):
raise SelfHostingLoopError(
"target_folder_path {!r} is not this instance's own checkout ({!r}) -- "
"the self-hosting scenario requires self-target".format(target_folder_path, paths.repo_root())
)
task_result = scheduler.run_goal_to_completion(db_path, target_folder_path, goal_text)
if task_result.get("final_status") != "done":
return {"stage": "task_lifecycle", "outcome": task_result.get("final_status"), "task_id": task_result.get("task_id")}
conn = schema.connect(db_path)
try:
bug = conn.execute(
"SELECT bug_id FROM bug_record WHERE task_id = ? AND root_cause_category = 'core_engine' "
"ORDER BY rowid DESC LIMIT 1",
(task_result["task_id"],),
).fetchone()
if bug is None:
return {"stage": "classification", "outcome": "not_core_engine", "task_id": task_result["task_id"]}
merge_commit = deploy_stage._current_commit(target_folder_path)
staged = deploy_stage.stage_deployment(
conn, merge_commit, repo_root=target_folder_path, prod_db_path=prod_db_path
)
health = deploy_health_check.run_health_check(
conn, staged["deployment_id"], staged["candidate_pid"], staged["candidate_port"]
)
if health["status"] != "health_check_passed":
candidate_db_path = conn.execute(
"SELECT candidate_db_path FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],)
).fetchone()[0]
deploy_rollback.rollback_path_a(
conn, staged["deployment_id"], staged["candidate_pid"], candidate_db_path,
merge_commit, repo_root=target_folder_path,
description="health check failed: {}".format(health["signals"]),
)
return {"stage": "health_check", "outcome": "rolled_back_path_a", "deployment_id": staged["deployment_id"]}
cutover_result = deploy_cutover.cutover(
conn, staged["deployment_id"], live_pid=None, candidate_pid=staged["candidate_pid"],
candidate_instance_dir=staged["candidate_instance_dir"], prod_db_path=prod_db_path,
)
project_id = conn.execute(
"SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
).fetchone()[0]
cutover_at = conn.execute(
"SELECT cutover_at FROM deployment WHERE deployment_id = ?", (staged["deployment_id"],)
).fetchone()[0]
baseline = deploy_monitor.compute_baseline_validator_approval_rate(conn, project_id, cutover_at)
baseline_rate = baseline[0] if baseline else None
cycles = 0
while True:
cycles += 1
cycle_result = deploy_monitor.run_monitor_cycle(
conn, staged["deployment_id"], cutover_result["live_pid"], cutover_result["live_port"],
project_id, baseline_rate, cutover_at, now=now,
)
if cycle_result["status"] == "monitoring_window_clean":
conn.execute(
"UPDATE deployment SET status = 'finalized' WHERE deployment_id = ?", (staged["deployment_id"],)
)
conn.commit()
return {"stage": "monitor", "outcome": "finalized", "deployment_id": staged["deployment_id"]}
if cycle_result["status"] == "anomaly_detected":
rb = deploy_rollback.rollback_path_b(
conn, staged["deployment_id"], cutover_result["live_pid"],
_previous_instance_dir(conn, staged["deployment_id"]),
prod_db_path, merge_commit, repo_root=target_folder_path,
description="post-cutover anomaly: {}".format(cycle_result["latest_signal"]),
)
outcome = "rolled_back_path_b" if rb["status"] == "rolled_back" else "rollback_failed"
return {"stage": "monitor", "outcome": outcome, "deployment_id": staged["deployment_id"]}
if monitor_timeout_cycles is not None and cycles >= monitor_timeout_cycles:
return {"stage": "monitor", "outcome": "timed_out_still_monitoring", "deployment_id": staged["deployment_id"]}
sleep(poll_interval)
finally:
conn.close()
def _previous_instance_dir(conn, deployment_id):
return conn.execute(
"SELECT previous_instance_dir FROM deployment WHERE deployment_id = ?", (deployment_id,)
).fetchone()[0]