host
9e56c1d266
Server writes goal + monotonic goal_version + task_goal_history in one transaction (CAS on version, approve_merge only, new-semantic goal tasks only, terminal/merged 409, same-goal noop). Worker mirror thread (30s) and breakpoint-resume claim hook write the update as a signal file next to host-state.json; host run_goal_loop applies it at round boundaries (top of round and before the declaration check, so a declaration made for the old goal in a turn during which the goal changed is voided), with an explicit old->new notice, same thread/worktree, rounds/usage/ budget/failed counters untouched. handler method accounting is a contract update: 239 total / 223 mixins / 16 in server.py. (cherry picked from commit 6841f287219d2b36cfbb8803642344b84dae4377) Reconcile-port: handler_lifecycle.py contents redirected — _update_goal_task method + _goal_update_reject/_goal_signal_status pure functions + goal_live_update import into cloud/server.py (target keeps 6-mixin split); _task_to_dict goal fields into cloud/handler_support.py; migrate columns into cloud/server.py migrate_tasks_table; test patch target handler_lifecycle→server; test_handler_split_conformance.py kept at target version, to be re-pinned for the unified line in a final commit.
195 lines
7.9 KiB
Python
195 lines
7.9 KiB
Python
"""Handler 共享 helper：被多个 handler 分组与留守代码共用。
|
||
|
||
由 cloud/server.py 拆出，行为与原来逐字一致；server.py 以
|
||
重导出保留旧导入路径（见 server.py 的 handler_support 导入块）。
|
||
本模块不 import cloud.server（避免循环），不直连 sqlite3，
|
||
不 import psycopg；存储一律走 cloud.db / 各 Store。
|
||
"""
|
||
|
||
from cloud.auth import AuthStore
|
||
from cloud.coordinator import parse_parts as _coord_parse_parts
|
||
from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups
|
||
from cloud.quota import default_quota_usd_micros as _quota_default
|
||
from cloud.peer_messages import row_get as _row_get
|
||
from contextlib import contextmanager
|
||
import json
|
||
import os
|
||
import time
|
||
|
||
|
||
def _env_int(name, default):
|
||
try:
|
||
value = int(os.environ.get(name, "") or default)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
return value if value > 0 else default
|
||
|
||
|
||
def _ensure_followups_table(conn):
|
||
"""幂等建后续消息表：委托给 peer_messages（两边同一份 DDL，
|
||
server init_db、各 followup 接口与 worker 投递路径共用）。
|
||
老库第一次跑到这里时自动补表补列，不丢数据。DOUBLE
|
||
PRECISION 在 SQLite 与 Postgres 下都可用（SQLite 按 NUMERIC
|
||
亲和处理）。"""
|
||
_peer_ensure_followups(conn)
|
||
|
||
|
||
def _ensure_quotas_table(conn):
|
||
"""幂等建配额表：server init_db 与配额查询都可调它，老库
|
||
第一次跑到这里时自动补表，不丢数据。quota_usd_micros 为
|
||
NULL 表示不限额（与 0 区分：0 是一分钱也不能花）；金额
|
||
单位与 cost_records 一致，都是 USD 微单位（1 美元 =
|
||
1_000_000），避免浮点。"""
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS user_quotas("
|
||
"user_id TEXT PRIMARY KEY, "
|
||
"quota_usd_micros BIGINT, "
|
||
"updated_at DOUBLE PRECISION)")
|
||
|
||
|
||
def _default_quota_usd_micros():
|
||
"""OAuth 新用户建号时写入的默认配额（微美元）。
|
||
|
||
无配额行 = 不限额，所以 OAuth 自动建号过去等于无限额入驻。
|
||
新号现在建号即写一行默认配额（老用户无配额行不受影响，
|
||
行为不变）。环境变量 CLOUD_DEFAULT_QUOTA_USD_MICROS 可
|
||
覆盖，默认 10_000_000（$10/月）；非法值按默认计。
|
||
解析收敛到 cloud/quota.py（别名保留）。
|
||
"""
|
||
return _quota_default()
|
||
|
||
|
||
def _apply_default_quota(conn, user_id):
|
||
"""给用户写默认配额行（已存在则不动，幂等）。
|
||
|
||
INSERT OR IGNORE 保证重复回调不覆盖管理员后来改过的额度。
|
||
返回 True 表示本次写了新行（调用方据此记 quota_default
|
||
审计）。P0 安全批后：OAuth 新建、绑定、已存在用户登录
|
||
三条分支都调它补行，不再只给新建用户补。
|
||
"""
|
||
_ensure_quotas_table(conn)
|
||
cur = conn.execute(
|
||
"INSERT OR IGNORE INTO user_quotas(user_id, quota_usd_micros,"
|
||
" updated_at) VALUES(?, ?, ?)",
|
||
(user_id, _default_quota_usd_micros(), time.time()))
|
||
try:
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
return cur.rowcount == 1
|
||
|
||
|
||
def _global_admin_ids():
|
||
"""真全局管理员清单：环境变量 CODEX_GLOBAL_ADMINS。
|
||
|
||
逗号分隔的用户 id 列表，空白项忽略。未配置或为空时返回
|
||
空集——此时 admin 端点一律 fail-closed，绝不回落到项目
|
||
角色推导（P0 安全批前“任一项目 maintainer 即全局管理”
|
||
的推导已被废掉，见 _is_global_admin）。
|
||
"""
|
||
raw = os.environ.get("CODEX_GLOBAL_ADMINS") or ""
|
||
return {part.strip() for part in raw.split(",") if part.strip()}
|
||
|
||
|
||
def _environment_to_dict(row):
|
||
return {
|
||
"id": row["id"],
|
||
"project_id": row["project_id"],
|
||
"name": row["name"],
|
||
"branch": _row_get(row, "branch"),
|
||
"executor": _row_get(row, "executor"),
|
||
"status": row["status"],
|
||
"error_text": _row_get(row, "error_text"),
|
||
"repo_url": _row_get(row, "repo_url"),
|
||
"created_by": row["created_by"],
|
||
"created_at": row["created_at"],
|
||
"updated_at": row["updated_at"],
|
||
}
|
||
|
||
|
||
@contextmanager
|
||
def _auth_store(db_path):
|
||
"""每请求一个 AuthStore 实例（自带新连接），用完即关闭。"""
|
||
store = AuthStore(db_path)
|
||
try:
|
||
yield store
|
||
finally:
|
||
try:
|
||
store.conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _task_to_dict(row, attachments=None, total_tokens=None):
|
||
depends_on = row["depends_on"]
|
||
try:
|
||
depends_on = json.loads(depends_on) if depends_on else []
|
||
except (TypeError, ValueError):
|
||
depends_on = []
|
||
if not isinstance(depends_on, list):
|
||
depends_on = [depends_on]
|
||
return {
|
||
"id": row["id"],
|
||
"project_id": row["project_id"],
|
||
"prompt": row["prompt"],
|
||
"status": row["status"],
|
||
"idempotency_key": row["idempotency_key"],
|
||
"depends_on": depends_on,
|
||
"priority": row["priority"],
|
||
"created_by": row["created_by"],
|
||
"created_at": row["created_at"],
|
||
"updated_at": row["updated_at"],
|
||
"last_heartbeat": row["last_heartbeat"],
|
||
"result_text": _row_get(row, "result_text"),
|
||
"result_full_available": bool(_row_get(row, "result_text_full")),
|
||
"error_text": _row_get(row, "error_text"),
|
||
"actual_model": _row_get(row, "actual_model"),
|
||
"started_at": _row_get(row, "started_at"),
|
||
"finished_at": _row_get(row, "finished_at"),
|
||
"child_pid": _row_get(row, "child_pid"),
|
||
"executor": _row_get(row, "executor"),
|
||
"pr_url": _row_get(row, "pr_url"),
|
||
"pr_number": _row_get(row, "pr_number"),
|
||
"goal": _row_get(row, "goal"),
|
||
"verify_command": _row_get(row, "verify_command"),
|
||
"goal_max_rounds": _row_get(row, "goal_max_rounds"),
|
||
"goal_mode": _row_get(row, "goal_mode") or 0,
|
||
# 运行中改 goal：goal_version 是已提交版本（服务端写库），
|
||
# applied_goal_version 是 host 循环已应用版本（镜像回写），
|
||
# 两者不等时说明信号在途或未到轮次边界，绝不能冒充已应用。
|
||
"goal_version": _row_get(row, "goal_version") or 0,
|
||
"applied_goal_version": _row_get(row, "applied_goal_version") or 0,
|
||
"attachments": list(attachments or []),
|
||
"resume_message": _row_get(row, "resume_message"),
|
||
"worktree_host": _row_get(row, "worktree_host"),
|
||
"claimed_by_worker": _row_get(row, "claimed_by_worker"),
|
||
"model": _row_get(row, "model"),
|
||
"effort": _row_get(row, "effort"),
|
||
"child_model": _row_get(row, "child_model"),
|
||
"child_effort": _row_get(row, "child_effort"),
|
||
"approval": _row_get(row, "approval"),
|
||
"plan": _row_get(row, "plan") or 0,
|
||
"forked_from": _row_get(row, "forked_from"),
|
||
"environment_id": _row_get(row, "environment_id"),
|
||
"turn_id": _row_get(row, "turn_id"),
|
||
"root_turn_id": _row_get(row, "root_turn_id"),
|
||
"coordinator_parts": (
|
||
_coord_parse_parts(_row_get(row, "coordinator_parts"))
|
||
if _row_get(row, "coordinator_parts") else None),
|
||
"parent_task_id": _row_get(row, "parent_task_id"),
|
||
"coordinator_part": _row_get(row, "coordinator_part"),
|
||
"goal_resume_pending": _row_get(row, "goal_resume_pending") or 0,
|
||
"paused_by_user": _row_get(row, "paused_by_user") or 0,
|
||
"interrupted_reason": _row_get(row, "interrupted_reason"),
|
||
"auto_resume_count": _row_get(row, "auto_resume_count") or 0,
|
||
"no_time_limit": _row_get(row, "no_time_limit") or 0,
|
||
# 列表接口由 _task_tokens_map 一次聚合传入；其它调用方
|
||
# 不传时为 None（前端显示占位），绝不把“未算”写成 0。
|
||
"total_tokens": total_tokens,
|
||
}
|
||
|
||
|
||
MAX_PROJECTS_PER_USER = _env_int("CODEX_MAX_PROJECTS_PER_USER", 50)
|
||
MAX_PROJECT_ID_LEN = _env_int("CODEX_MAX_PROJECT_ID_LEN", 64)
|
||
MAX_PROJECT_NAME_LEN = _env_int("CODEX_MAX_PROJECT_NAME_LEN", 100)
|