diff --git a/examples/kernels/openearth/.gitignore b/examples/kernels/openearth/.gitignore new file mode 100644 index 00000000..82eb8328 --- /dev/null +++ b/examples/kernels/openearth/.gitignore @@ -0,0 +1,8 @@ +# OpenEarth SDK build inputs are private and stay local. XSkill receives only +# the bridge, public documentation, configuration example, wheel, and checksum. +sdk/ + +__pycache__/ +config.yaml +workspace/ +output/ diff --git a/examples/kernels/openearth/README.md b/examples/kernels/openearth/README.md new file mode 100644 index 00000000..22213cb2 --- /dev/null +++ b/examples/kernels/openearth/README.md @@ -0,0 +1,140 @@ +# OpenEarth Kernel + +这个目录是 OpenEarth 接入 XSkill Kernel API v2 的可交付桥接包。训练只有一个 SDK +入口 `train_skills(...)`，输入始终是 XSkill 轨迹，不再区分“真实数据训练接口”和“数据集 +任务训练接口”。 + +## 数据流 + +```text +ready TrajectoryResource + └─ atoms: AtomResource[] + ├─ user trajectory → atom.ux_score + └─ temp trajectory → OpenEarth oracle score + ↓ + ScoredAtomInput + ↓ + reflect / curate → SkillDraft + ↓ + context.publisher.submit(...) +``` + +在启用内部 benchmark 时，full rebuild 会先由 OpenEarth target harness 运行配置的 case， +把每个 rollout 转成一个平台格式 Markdown，记录 oracle score，再调用 +`context.trajectories.create_temp(...)`。临时轨迹进入 XSkill 拆分队列；拆成唯一 Atom +并变为 ready 后，在后续 Kernel 调用中进入上面的统一训练流。 + +桥接层仅提交 `atom_split_status == "ready"` 的轨迹。输入范围严格遵循 +`context.invocation`： + +- `changed_trajectory_ids` 非空时，只处理其中的 ready 轨迹； +- changed 为空且 `full_rebuild=True` 时，处理全部 ready 轨迹； +- changed 为空且非 full rebuild 时不蒸馏轨迹、不调用 SDK 训练入口，但仍检查待发布队列。 + +SDK 将每条轨迹展开成 atom，并以 `#` 作为稳定证据 ID： + +- 真实用户轨迹（`source="user"`）直接读取每个 `atom.ux_score`； +- 评测轨迹（`source="temp"`）忽略 atom 自身分数，读取 OpenEarth 在 + `context.workspace` 中按 `trajectory_id` 保存的 oracle 分数； +- `pending` 和 `updated` 轨迹不会进入本轮训练； +- UX 7–10 为成功，1–5 为失败，6 或未评分暂缓； +- 当前链路没有 Gate 阶段。 + +已有 XSkill Skill 会作为只读 main 快照传给 SDK。显式声明 `level` 的 Skill 直接进入 +对应层；没有 `level` 的 Skill 由 OpenEarth 在一次批量 LLM 调用中分类为 +`planning`、`functional` 或 `unclassified`。分类结果按 `name + main_commit_sha` +缓存在 workspace，不会写回原 Skill；低于 0.7 置信度的结果保持未分类。 + +## OpenEarth 处理流程 + +一次 Kernel 运行大致分为以下阶段： + +1. **可选 benchmark 生产**：启用 benchmark 的 full rebuild 先运行尚未登记的 case， + 计算 oracle score，并创建 pending 临时轨迹；这些轨迹等待 XSkill 拆分，不在当轮训练。 +2. **选择输入**：优先处理本轮变化且已经完成 atom 拆分的轨迹；手动全量运行时读取全部 + ready 轨迹；没有变化且不是全量重建时本轮不做任何训练。 +3. **同步已有 Skill**：把 XSkill main Skill 的完整只读快照放入 OpenEarth workspace， + 并对没有 `level` 的新增或变化 Skill 做一次批量分类。 +4. **构造训练证据**：把每条轨迹展开成 atom，真实数据使用 `atom.ux_score`，临时评测 + 数据使用已记录的 OpenEarth oracle score。增量运行通过稳定证据 ID 和内容签名跳过 + 历史未变化 atom；全量重建会重新蒸馏历史 atom，但同一批内仍按 `atom_id` 去重， + 同 ID 不同内容或评分会直接报冲突。 +5. **划分反思通道**：UX 7–10 的成功 atom 用于提炼可复用 Planning Skill；UX 1–5 + 的失败 atom 先检索已有 Planning Skill，找不到时生成临时诊断计划，再反思产生 + Functional 修复候选。 +6. **整理候选**：对本轮 Functional 候选做一次全局 curation，可保留、修改、合并或 + 丢弃候选，同时校验其 atom 来源。 +7. **生成草稿**：候选名称与已有 Skill 完全相同时生成 update draft，并保留原 bundle； + 否则生成 create draft。 +8. **交给 XSkill 发布**：SDK 只返回 `SkillDraft`。Kernel 再通过 + `context.publisher.submit(...)` 发布；已有 active staging 的 Skill 会进入 OpenEarth + 待发布队列。 + +内部 benchmark 只负责产生训练证据，不执行 candidate rollout 或 OpenEarth Gate。 +XSkill 自己的 staging/canary 发布机制不受影响。 + +全量重建仍以当前 main Skill 作为只读反思上下文，并复用 +`name + main_commit_sha` 分类缓存，因此不会重新分类未变化的 Skill。重建只生成并提交 +本轮草稿，不会删除没有重新生成的旧 Skill。 + +## staging 发布队列 + +XSkill 同一时刻只允许同名 Skill 存在一个 active staging。OpenEarth 不再丢弃因此无法 +提交的新草稿，而是在 `context.workspace/openearth-publication-queue.json` 中为每个 +Skill 保留一个最新 pending draft： + +- active staging 仍存在时继续等待；新 draft 覆盖同名旧 pending draft，避免陈旧版本 + 无限堆积； +- staging 被拒绝、main 未变化时，pending draft 使用原 base commit 直接提交； +- staging 晋升、main 已变化时，先保留最新 main 的 provider 元数据和 bundle 文件， + 再叠加 pending draft 的 OpenEarth 字段和正文，最后以新 main SHA 提交； +- 每次 scheduled/manual 调用都会先执行一次队列 tick，即使本轮没有 changed trajectory； +- 只有 Publisher 成功接受草稿后才从队列移除，发布竞态则重新读取 main/staging 状态。 + +队列只管理尚未进入 XSkill staging 的下一版本；已进入 staging 的版本仍完全由 XSkill +灰度、晋升或拒绝。 + +## 安装 + +SDK 源码位于本地 `sdk/`，由本目录的 `.gitignore` 排除，不会提交到 XSkill 远程仓库。 +仓库交付的是构建后的 wheel： + +```bash +python -m pip install \ + examples/kernels/openearth/wheels/openearth_skill_sdk-0.9.0-py3-none-any.whl +``` + +然后复制 Kernel 目录并创建私有配置： + +```bash +mkdir -p "$HOME/.xskill/kernels" +cp -R examples/kernels/openearth "$HOME/.xskill/kernels/openearth" +cp "$HOME/.xskill/kernels/openearth/config.yaml.example" \ + "$HOME/.xskill/kernels/openearth/config.yaml" +``` + +在 `~/.xskill/config.yaml` 中选择： + +```yaml +kernel: + kernels_path: ~/.xskill/kernels + kernel_id: openearth +``` + +详细 SDK 和评测接入方式见 [docs/sdk-usage.md](docs/sdk-usage.md)。 + +## 构建私有 SDK + +在本地源码目录执行： + +```bash +cd examples/kernels/openearth/sdk +python -m pip wheel \ + --no-deps \ + --no-build-isolation \ + --wheel-dir ../wheels \ + . +``` + +每次重建后同步更新 `SHA256SUMS`。wheel 包含 benchmark dataset/environment/target +harness 和运行时代码，但不包含 Gate、组合 experiment 或 SDK 测试。 diff --git a/examples/kernels/openearth/SHA256SUMS b/examples/kernels/openearth/SHA256SUMS new file mode 100644 index 00000000..423b209a --- /dev/null +++ b/examples/kernels/openearth/SHA256SUMS @@ -0,0 +1 @@ +daa898d4dbc1d64729d43d1368736dedd66fa049448d61e75c70643b85b1e223 wheels/openearth_skill_sdk-0.9.0-py3-none-any.whl diff --git a/examples/kernels/openearth/config.yaml.example b/examples/kernels/openearth/config.yaml.example index 5db88275..dee77b15 100644 --- a/examples/kernels/openearth/config.yaml.example +++ b/examples/kernels/openearth/config.yaml.example @@ -1,4 +1,30 @@ -# This file belongs to the OpenEarth kernel. XSkill never parses or edits it. -model: YOUR_OPENEARTH_MODEL -api_key: YOUR_OPENEARTH_KEY +# Private OpenEarth SDK configuration. XSkill treats this file as opaque. +reflect: + # OpenEarth currently invokes the configured model through OpenCode. + base_url: opencode + model: YOUR_OPENEARTH_MODEL + binary: opencode + timeout: 600 + # Optional custom reflection strategy Markdown: + # strategy: /absolute/path/to/strategy.md +# Optional OpenEarth-internal benchmark producer. It runs only on a +# full_rebuild invocation. Empty/disabled configuration keeps live-data-only +# training behavior. +benchmark: + enabled: false + # dataset_dir: /absolute/path/to/officeqa + env: officeqa # officeqa | spreadsheet | livemath + split: train + model: YOUR_TARGET_AGENT_MODEL + binary: opencode + agent_timeout: 900 + parallel: 1 + # Stable cases already registered in the workspace are not rerun. + # n_cases: 10 + # Change this value to intentionally produce a fresh rollout for each case. + # sample_id: default + planning_route_k: 2 + functional_route_k: 2 + # OfficeQA optional document roots: + # officeqa_docs_dirs: /absolute/path/to/parsed/docs diff --git a/examples/kernels/openearth/docs/sdk-usage.md b/examples/kernels/openearth/docs/sdk-usage.md new file mode 100644 index 00000000..a21fcf18 --- /dev/null +++ b/examples/kernels/openearth/docs/sdk-usage.md @@ -0,0 +1,251 @@ +# OpenEarth SDK 接口 + +## 统一训练入口 + +XSkill 桥接层调用唯一入口： + +```python +from openearth_skill_sdk import ExistingSkillInput, train_skills + +result = train_skills( + config_path=context.config_path, + workspace=context.workspace, + trajectories=ready_trajectories, + existing_skills=existing_skills, + run_id=context.run_id, + full_rebuild=context.invocation.full_rebuild, +) +``` + +`trajectories` 可以是 `TrajectoryResource` iterable，也可以是提供 `iter()` 的 +`TrajectoryReader`。生产桥接层会先筛选 `atom_split_status == "ready"` 的资源。 + +SDK 内部将 `trajectory.atoms` 转成 OpenEarth 自己定义、与 XSkill provider 解耦的 +`ScoredAtomInput`： + +```python +@dataclass(frozen=True) +class ScoredAtomInput: + atom_id: str + evidence_id: str + parent_trajectory_id: str + content: str + ux_score: int | None + intent: str = "" + summary: str = "" + used_skills: tuple[str, ...] = () + score_source: Literal["xskill", "oracle"] = "xskill" + metadata: Mapping[str, object] = field(default_factory=dict) +``` + +普通接入方不需要自行构造它；它也是 SDK 的公开中间契约，便于脱离 XSkill +`TrajectoryResource` 做单元测试或其他 provider 适配。 + +## 增量与全量重建 + +Kernel 按下面的三态契约选择 SDK 输入： + +| `changed_trajectory_ids` | `full_rebuild` | 本轮行为 | +| --- | --- | --- | +| 非空 | 任意值 | 只选择 changed 中的 ready 轨迹 | +| 空 | `True` | 选择全部 ready 轨迹 | +| 空 | `False` | 不训练轨迹；只执行待发布队列 tick | + +changed 始终拥有输入范围的优先级。`full_rebuild=True` 会传给 SDK，使所选轨迹中的历史 +atom 绕过跨运行签名去重并重新进入蒸馏。无论是否重建，同一训练批次都按稳定 +`atom_id` 去重：ID 和签名都相同只处理一次；同一 ID 对应不同证据、内容或评分时抛出 +冲突错误，避免重复证据产生不一致结果。 + +全量重建仍读取当前 main Skill 作为反思上下文，但 `name + main_commit_sha` 未变化的 +Skill 会命中原分类缓存，不再次调用分类 LLM。SDK 和 Kernel 都不会因为某个旧 Skill +没有在本轮重新生成就将它删除。 + +队列为空时，上表最后一行不会读取轨迹或 Skill，也不会调用 SDK 训练入口；队列非空时 +只读取对应 Skill 的当前 main/staging 状态，不执行轨迹反思或 LLM 调用。 + +## 已有 Skill 的分层 + +Kernel 将 XSkill 的所有已有 main Skill 作为 `ExistingSkillInput` 传入，包括完整 +`SKILL.md`、其他 bundle 文本文件和 `main_commit_sha`。 + +SDK 按以下顺序决定反思层级： + +1. 顶层 `level` 或 `metadata.level` 已明确声明时直接沿用，不调用 LLM； +2. 没有 `level` 时，将本轮所有新增或版本变化的未分类 Skill 放入一次批量 LLM 调用； +3. 分类输出只能是 `planning`、`functional` 或 `unclassified`； +4. `planning` / `functional` 的置信度至少为 0.7 才生效，否则保持 + `unclassified`； +5. 结果按 Skill 名称和 provider `version_token` 缓存。XSkill 中该 token 是 + `main_commit_sha`，main 未变化时不重复调用。 + +缓存位于： + +```text +/openearth-skill-level-classifications.json +``` + +这是 OpenEarth 私有映射，不会修改 XSkill 原 Skill 的 frontmatter。未分类 Skill 不再 +默认当作 Functional，也不会进入 Planning/Functional 反思上下文。候选名称与已有 Skill +名称完全相同时，SDK 才生成 update draft。 + +## 训练流水线 + +`train_skills` 内部按以下顺序处理： + +```text +ready trajectories + → 展开并标准化 ScoredAtomInput + → 批内按 atom_id 去重 + → 增量时使用 evidence_id + 内容/评分签名过滤历史未变化 atom + （full rebuild 绕过这一层） + → 按 UX 分数划分 success / failure / deferred + → success：提炼 Planning 候选 + → failure：检索 Planning 上下文或生成临时诊断计划 + 再生成 Functional 修复候选 + → 全局整理 Functional 候选 + → 组装 create/update SkillDraft +``` + +具体行为： + +- **证据去重**：批内先按 `atom_id` 去重并拒绝冲突副本。跨运行状态使用 + `evidence_id`（轨迹资源 ID 与 atom ID）及内容/评分签名；增量时内容、评分、评分来源 + 或摘要发生变化才重新进入反思，全量重建则重新处理历史证据。 +- **Planning 通道**：成功 atom 用来提炼跨步骤、可复用的规划；本轮新 Planning 候选与 + 已有 Planning Skill 一起构成失败通道的检索库。 +- **Functional 通道**：每个失败 atom 先检索相关 Planning Skill；没有匹配项时只生成 + 本轮使用的临时诊断计划，不把它作为 Skill 发布。随后生成局部操作或修复候选。 +- **Curation**：对本轮所有 Functional 候选做一次全局整理，可以保留、编辑、合并或 + 丢弃，但必须保留可验证的来源 atom。 +- **草稿判定**：候选名称与已有 Skill 名称精确一致时为 update，否则为 create。update + 会继承已有 frontmatter 中的非优化字段和其他 bundle 文件。 +- **状态边界**：轨迹证据、分类缓存、处理签名及候选保存在 + `context.workspace`；SDK 不直接修改 XSkill Skill 仓库。 +- **发布边界**：SDK 返回草稿后，由 Kernel 调用 XSkill Publisher。当前不执行 + OpenEarth Gate，但 XSkill 仍按自身 staging/canary 规则处理已有 Skill 的更新。 + +## 真实用户数据 + +真实数据无需额外评分接口。SDK 对 ready 轨迹逐个展开 atom，直接读取 XSkill 已提供的 +`atom.ux_score`： + +```python +for trajectory in ready_trajectories: + for atom in trajectory.atoms: + # atom.ux_score: 1..10 或 None + ... +``` + +OpenEarth 不调用 XSkill 内部 UX scorer，也不读取轨迹级分数，因为 +`TrajectoryResource` 没有轨迹级 `ux_score`。 + +## 评测数据和 oracle 分数 + +启用 `benchmark.enabled` 后，Kernel 在 full rebuild 中调用 OpenEarth +`run_benchmark(...)`。harness 从 `benchmark.dataset_dir` 加载 case，运行 target agent +并用环境 oracle 评分，再使用同一个稳定 `trajectory_id` 依次写分和创建临时轨迹。 +Gate 不会运行。 + +```yaml +benchmark: + enabled: true + dataset_dir: /absolute/path/to/officeqa + env: officeqa + split: train + model: deepseek/deepseek-v4-flash + binary: opencode + agent_timeout: 900 + parallel: 1 + n_cases: 10 + # 修改 sample_id 可为同一批 case 主动生成一组新 rollout + sample_id: default + officeqa_docs_dirs: /absolute/path/to/parsed/docs +``` + +Kernel 的登记动作等价于： + +```python +from openearth_skill_sdk import record_oracle_score + +trajectory_id = "traj_oe_case_001" + +record_oracle_score( + workspace=context.workspace, + trajectory_id=trajectory_id, + ux_score=1, # 1..10 + case_id="case-001", + metadata={"suite": "smoke"}, +) + +temp = context.trajectories.create_temp( + markdown=( + "## User\n\n" + "Benchmark task.\n\n" + "## Assistant\n\n" + "Rollout result.\n" + ), + trajectory_id=trajectory_id, +) + +assert temp.source == "temp" +assert temp.atom_split_status == "pending" +``` + +`create_temp` 只登记待拆分轨迹；不要轮询。平台完成 atom 拆分后，它会在后续 Kernel +调用中以 `ready` 资源出现，此时 SDK 按 `trajectory.trajectory_id` 找到已保存的 oracle +分数并训练。 + +benchmark 状态保存在 +`/openearth-benchmark-state.json`。轨迹 ID 由数据路径、环境、split、 +case 内容、target model 和 `sample_id` 生成；相同 case 登记成功后不会在后续 full +rebuild 中重复运行。要主动生成一组新 rollout，可以修改 `benchmark.sample_id`。 + +当前恢复的 harness 支持 `officeqa`、`spreadsheet` 和 `livemath`。其任务工作目录和 +OpenCode 隔离数据位于 `/openearth-benchmark-runs`，完成后清理临时 +目录；正式训练证据仍由 XSkill temp trajectory 保存。 + +一个 oracle 分数必须只对应一个 atom，因此评测 Markdown 应表达一个完整的 +User/Assistant rollout。若 ready 临时轨迹被拆成多个 atom，SDK 会报错，避免把一个 case +级分数错误地复制到多段证据。 + +oracle 分数保存在： + +```text +/openearth-oracle-scores.json +``` + +这是 OpenEarth 私有状态，不写入 XSkill 的 trajectory sidecar，也不会调用或覆盖 +`atom.ux_score`。 + +## 输出与发布 + +`train_skills` 返回 `TrainingResult`，其中包含： + +- `drafts`：完整 Skill 草稿； +- `processed_trajectory_ids`：已消费的 XSkill 轨迹资源 ID； +- `processed_atom_ids`：已消费的稳定 atom 证据 ID； +- `metrics`：成功、失败、暂缓和评分来源计数； +- `candidate_dir`：OpenEarth workspace 中的候选目录。 + +SDK 不直接写 XSkill Skill 仓库。`kernel.py` 通过 +`context.publisher.submit(SkillSubmission(...))` 发布。本版本暂时不执行 Gate。 + +### active staging 与排队 + +已有 Skill 存在 active staging 时，Kernel 将草稿写入： + +```text +/openearth-publication-queue.json +``` + +这是按 Skill 名称组织的 latest-wins 队列：一个 Skill 最多保留一个尚未提交的 pending +draft，新 draft 会替换旧 pending draft。每次 Kernel 调用先检查队列： + +1. staging 仍存在：继续等待； +2. staging 被拒绝且 main SHA 未变化：按原 base commit 提交； +3. staging 晋升且 main SHA 已变化：确定性 rebase 到最新 main，再提交； +4. Publisher 成功后删除 pending；竞态导致 staging/main 改变时刷新状态后等待或重试。 + +rebase 不调用 LLM，也不是把旧 bundle 整体覆盖到新 main。它保留最新 main 的 +provider-owned frontmatter 和附件，应用 pending draft 的 description、OpenEarth +optimizer 字段和正文，并把 `base_commit_sha` 更新为最新 main SHA。 diff --git a/examples/kernels/openearth/kernel.py b/examples/kernels/openearth/kernel.py index 2a049337..d2c6676f 100644 --- a/examples/kernels/openearth/kernel.py +++ b/examples/kernels/openearth/kernel.py @@ -1,30 +1,29 @@ -"""Illustrative local bridge for an SDK package named ``openearth``. - -Copy this directory to ``~/.xskill/kernels/openearth`` and replace the marked -adapter call with the real SDK API. XSkill imports this bridge as trusted code. - -Trajectory / atom feed notes for OpenEarth adapters: - -- Prefer ``context.invocation.changed_trajectory_ids`` (ready-only). Read - ``traj.atoms`` and dedupe by ``atom_id`` in ``context.workspace``. -- Mother trajectory ``read_text()`` is always available; there is no - trajectory-level ``ux_score`` (scores live on atoms / Skill versions). -- For algorithm-owned rollouts, convert OE harness markdown to **platform** - style (``## User`` / ``## Assistant`` / …) on the OE side, then call - ``context.trajectories.create_temp(markdown, trajectory_id=...)``. - ``trajectory_id`` must match ``traj_[a-z0-9]...``. The returned resource is - ``source="temp"`` / ``atom_split_status="pending"``; do **not** spin-wait — - atoms arrive later via the ready feed. +"""XSkill bridge for the private OpenEarth Skill SDK wheel. + +The SDK receives only ready trajectory resources. It expands their Atom +views, using XSkill ``atom.ux_score`` for user trajectories and the OpenEarth +oracle score store for kernel-owned temporary benchmark trajectories. """ from __future__ import annotations +import json +from pathlib import Path + try: - import openearth + from openearth_skill_sdk import ( + ExistingSkillInput, + SkillDraft, + __version__, + rebase_skill_draft, + record_oracle_score, + run_benchmark, + train_skills, + ) except ImportError as exc: raise ImportError( - "OpenEarth kernel requires its SDK package; install it in the same " - "Python environment as xskill" + "OpenEarth kernel requires openearth-skill-sdk; install the supplied " + "wheel in the same Python environment as xskill" ) from exc from xskill.kernels import ( @@ -35,71 +34,330 @@ ) +_QUEUE_SCHEMA = 1 +_QUEUE_FILENAME = "openearth-publication-queue.json" + + +def _skill_snapshot(skill) -> ExistingSkillInput: + files = { + relative_path: skill.read_text(relative_path) + for relative_path in skill.list_files() + if relative_path != "SKILL.md" + } + return ExistingSkillInput( + name=skill.name, + skill_md=skill.read_text(), + files=files, + version_token=skill.main_commit_sha, + ) + + +def _existing_skills(context) -> tuple[ExistingSkillInput, ...]: + return tuple(_skill_snapshot(skill) for skill in context.skills.list()) + + +def _queue_path(context) -> Path: + return Path(context.workspace) / _QUEUE_FILENAME + + +def _load_queue(context) -> dict: + path = _queue_path(context) + if not path.is_file(): + return {"schema": _QUEUE_SCHEMA, "pending": {}} + value = json.loads(path.read_text(encoding="utf-8")) + if ( + not isinstance(value, dict) + or value.get("schema") != _QUEUE_SCHEMA + or not isinstance(value.get("pending"), dict) + ): + raise ValueError(f"invalid OpenEarth publication queue: {path}") + return value + + +def _save_queue(context, queue: dict) -> None: + path = _queue_path(context) + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.tmp") + temporary.write_text( + json.dumps(queue, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + temporary.replace(path) + + +def _draft_to_queue_entry(draft, *, run_id: str) -> dict: + return { + "name": draft.name, + "skill_md": draft.skill_md, + "files": dict(draft.files), + "source_trajectory_ids": list(draft.source_trajectory_ids), + "action": draft.action, + "base_version_token": draft.base_version_token, + "generated_run_id": run_id, + } + + +def _draft_from_queue_entry(name: str, entry: dict) -> SkillDraft: + if not isinstance(entry, dict) or entry.get("name") != name: + raise ValueError(f"invalid queued OpenEarth Skill draft: {name!r}") + files = entry.get("files", {}) + source_ids = entry.get("source_trajectory_ids", []) + if not isinstance(files, dict) or not isinstance(source_ids, list): + raise ValueError(f"invalid queued OpenEarth Skill bundle: {name!r}") + return SkillDraft( + name=name, + skill_md=str(entry.get("skill_md") or ""), + files=files, + source_trajectory_ids=tuple(str(item) for item in source_ids), + action=str(entry.get("action") or "update"), + base_version_token=entry.get("base_version_token"), + ) + + +def _current_skill(context, name: str): + try: + return context.skills.get(name) + except KeyError: + return None + + +def _prepare_for_current_main(draft, current): + if current is None: + return draft, False + if draft.base_version_token == current.main_commit_sha: + return draft, False + return rebase_skill_draft(draft, _skill_snapshot(current)), True + + +def _submit(context, draft, current, *, origin_run_id: str): + prepared, rebased = _prepare_for_current_main(draft, current) + published = context.publisher.submit(SkillSubmission( + name=prepared.name, + skill_md=prepared.skill_md, + files=prepared.files, + source_trajectory_ids=prepared.source_trajectory_ids, + message=f"OpenEarth run {origin_run_id}", + base_commit_sha=( + current.main_commit_sha if current is not None else None + ), + )) + return published, rebased + + +def _submit_or_observe_busy(context, draft, current, *, origin_run_id: str): + """Submit once, refreshing main if canary changed during the attempt.""" + for attempt in range(2): + if current is not None and current.staging_commit_sha: + return None, False, True + current_token = ( + current.main_commit_sha if current is not None else None + ) + try: + published, rebased = _submit( + context, + draft, + current, + origin_run_id=origin_run_id, + ) + return published, rebased, False + except RuntimeError: + refreshed = _current_skill(context, draft.name) + if refreshed is not None and refreshed.staging_commit_sha: + return None, False, True + refreshed_token = ( + refreshed.main_commit_sha if refreshed is not None else None + ) + if attempt == 0 and refreshed_token != current_token: + current = refreshed + continue + raise + raise AssertionError("unreachable publication retry state") + + +def _drain_publication_queue(context, queue: dict): + submitted = [] + metrics = { + "queue_pending_before": len(queue["pending"]), + "queue_waiting": 0, + "queue_missing_skill": 0, + "queue_drained": 0, + "queue_rebased": 0, + } + for name in sorted(tuple(queue["pending"])): + entry = queue["pending"][name] + draft = _draft_from_queue_entry(name, entry) + current = _current_skill(context, name) + if current is None and draft.base_version_token is not None: + metrics["queue_missing_skill"] += 1 + continue + _published, rebased, busy = _submit_or_observe_busy( + context, + draft, + current, + origin_run_id=str(entry.get("generated_run_id") or context.run_id), + ) + if busy: + metrics["queue_waiting"] += 1 + continue + if rebased: + metrics["queue_rebased"] += 1 + submitted.append(name) + metrics["queue_drained"] += 1 + del queue["pending"][name] + _save_queue(context, queue) + return submitted, metrics + + +def _enqueue(context, queue: dict, draft) -> bool: + superseded = draft.name in queue["pending"] + queue["pending"][draft.name] = _draft_to_queue_entry( + draft, + run_id=context.run_id, + ) + _save_queue(context, queue) + return superseded + + class OpenEarthKernel(BaseKernel): metadata = KernelMetadata( id="openearth", name="OpenEarth", - version=getattr(openearth, "__version__", "unknown"), - description="OpenEarth SDK bridge example.", + version=__version__, + description="OpenEarth Atom-based Skill training bridge.", triggers=("scheduled", "manual"), api_version=2, ) def run(self, context, run_interval: int = 30) -> KernelRunResult: - # Prefer the ready-only changed feed when the host provides it. - # Offline distill / empty changed → walk all mother trajectories. + del run_interval + queue = _load_queue(context) + submitted, queue_metrics = _drain_publication_queue(context, queue) changed = tuple(context.invocation.changed_trajectory_ids) + full_rebuild = bool(context.invocation.full_rebuild) + existing_snapshots = None + benchmark_metrics = { + "benchmark_enabled": False, + "benchmark_selected": 0, + "benchmark_created": 0, + "benchmark_skipped": 0, + } + if full_rebuild: + existing_snapshots = _existing_skills(context) + + def register_benchmark_trajectory(rollout): + record_oracle_score( + workspace=context.workspace, + trajectory_id=rollout.trajectory_id, + ux_score=rollout.ux_score, + case_id=rollout.case_id, + metadata=rollout.metadata, + ) + context.trajectories.create_temp( + markdown=rollout.markdown, + trajectory_id=rollout.trajectory_id, + ) + + benchmark_result = run_benchmark( + config_path=context.config_path, + workspace=context.workspace, + existing_skills=existing_snapshots, + run_id=context.run_id, + on_trajectory=register_benchmark_trajectory, + ) + benchmark_metrics = dict(benchmark_result.metrics) if changed: - by_id = {item.id: item for item in context.trajectories.list()} - trajectories = [by_id[item_id] for item_id in changed if item_id in by_id] + changed_ids = set(changed) + selected = [ + trajectory + for trajectory in context.trajectories.list() + if trajectory.id in changed_ids + and trajectory.atom_split_status == "ready" + ] + elif full_rebuild: + selected = [ + trajectory + for trajectory in context.trajectories.list() + if trajectory.atom_split_status == "ready" + ] else: - trajectories = context.trajectories.list() - - # Illustrative create_temp path (adapter must convert markdown first): - # - # platform_md = openearth.to_platform_markdown(oe_rollout_md) - # temp = context.trajectories.create_temp( - # platform_md, - # trajectory_id="traj_oe_rollout_001", - # ) - # # temp.source == "temp"; temp.atom_split_status == "pending" - # # Do not poll — next ready feed will include atoms when split done. - # - # When consuming a ready trajectory: - # for atom in traj.atoms: # atom.atom_id / content / ux_score / used_skills - # ... - - # Adapt this single call to the real SDK. The SDK owns and parses its - # config; XSkill only passes the opaque path and standard inputs. - drafts = openearth.xskill_distill( # type: ignore[attr-defined] + return KernelRunResult( + submitted_skills=tuple(submitted), + metrics={ + "selected_trajectories": 0, + "full_rebuild": False, + "no_changes": True, + "generated_drafts": 0, + "published_drafts": len(submitted), + "queued_drafts": 0, + "queue_superseded": 0, + "queue_pending": len(queue["pending"]), + **queue_metrics, + **benchmark_metrics, + }, + notes=( + "No trajectory training input; OpenEarth only reconciled " + "its publication queue and did not invoke the SDK." + ), + ) + + result = train_skills( config_path=context.config_path, workspace=context.workspace, - trajectories=trajectories, - dataset_id=context.invocation.dataset_id, + trajectories=selected, + existing_skills=( + existing_snapshots + if existing_snapshots is not None + else _existing_skills(context) + ), + run_id=context.run_id, + full_rebuild=full_rebuild, ) - submitted = [] - consumed = [] - for draft in drafts: - base_commit = None - try: - base_commit = context.skills.get(draft.name).main_commit_sha - except KeyError: - pass - context.publisher.submit(SkillSubmission( - name=draft.name, - skill_md=draft.skill_md, - files=draft.files, - source_trajectory_ids=tuple(draft.trajectory_ids), - message=f"OpenEarth run {context.run_id}", - base_commit_sha=base_commit, - )) + queued = [] + superseded = [] + rebased_immediate = 0 + for draft in result.drafts: + current = _current_skill(context, draft.name) + _published, rebased, busy = _submit_or_observe_busy( + context, + draft, + current, + origin_run_id=context.run_id, + ) + if busy: + if _enqueue(context, queue, draft): + superseded.append(draft.name) + queued.append(draft.name) + continue + if rebased: + rebased_immediate += 1 submitted.append(draft.name) - consumed.extend(draft.trajectory_ids) + + metrics = { + **dict(result.metrics), + "selected_trajectories": len(selected), + "full_rebuild": full_rebuild, + "no_changes": False, + "processed_atoms": len(result.processed_atom_ids), + "generated_drafts": len(result.drafts), + "published_drafts": len(submitted), + "queued_drafts": len(queued), + "queue_superseded": len(superseded), + "queue_pending": len(queue["pending"]), + "queue_rebased_immediate": rebased_immediate, + **queue_metrics, + **benchmark_metrics, + } + if result.candidate_dir: + metrics["candidate_dir"] = result.candidate_dir return KernelRunResult( - processed_trajectory_ids=tuple(dict.fromkeys(consumed)), + processed_trajectory_ids=result.processed_trajectory_ids, submitted_skills=tuple(submitted), - metrics={"sdk_outputs": len(submitted)}, + metrics=metrics, + notes=( + "OpenEarth trained from ready trajectory Atom views; " + "XSkill owns publication and active staging drafts are queued." + ), ) diff --git a/examples/kernels/openearth/tests/test_delivery.py b/examples/kernels/openearth/tests/test_delivery.py new file mode 100644 index 00000000..d6a7a60b --- /dev/null +++ b/examples/kernels/openearth/tests/test_delivery.py @@ -0,0 +1,150 @@ +"""Checks for the source-private OpenEarth Kernel delivery bundle.""" + +from __future__ import annotations + +import hashlib +from pathlib import Path +from zipfile import ZipFile + + +OPENEARTH = Path(__file__).resolve().parents[1] +WHEEL = ( + OPENEARTH / "wheels" + / "openearth_skill_sdk-0.9.0-py3-none-any.whl" +) + + +def test_openearth_wheel_checksum_and_public_contents(): + expected = (OPENEARTH / "SHA256SUMS").read_text(encoding="utf-8").split()[0] + assert hashlib.sha256(WHEEL.read_bytes()).hexdigest() == expected + + with ZipFile(WHEEL) as archive: + names = set(archive.namelist()) + metadata = archive.read( + "openearth_skill_sdk-0.9.0.dist-info/METADATA" + ).decode("utf-8") + + assert "Version: 0.9.0" in metadata + assert "openearth_skill_sdk/xskill.py" in names + assert "openearth_skill_sdk/drafts.py" in names + assert "openearth_skill_sdk/benchmark.py" in names + assert "openearth_skill_sdk/environments.py" in names + assert "openearth_skill_sdk/target.py" in names + assert not any( + forbidden in name + for name in names + for forbidden in ("/gate.py", "/experiment.py", "/tests/") + ) + + +def test_openearth_delivery_documents_atom_score_sources(): + readme = (OPENEARTH / "README.md").read_text(encoding="utf-8") + usage = ( + OPENEARTH / "docs" / "sdk-usage.md" + ).read_text(encoding="utf-8") + + assert "atom.ux_score" in readme + assert "OpenEarth oracle score" in readme + assert "record_oracle_score" in usage + assert "context.trajectories.create_temp" in usage + assert "openearth-skill-level-classifications.json" in usage + assert "changed_trajectory_ids" in usage + assert "full_rebuild" in usage + assert "atom_id" in usage + assert "openearth-publication-queue.json" in usage + assert "latest-wins" in usage + assert "run_benchmark" in usage + assert "openearth-benchmark-state.json" in usage + + +def test_oracle_temp_multi_atom_does_not_raise(): + """Platform may split temp trajectories into N atoms; oracle is case-level.""" + import importlib + import sys + import types + from zipfile import ZipFile + + # Load xskill module from the vendored wheel without installing. + with ZipFile(WHEEL) as archive: + source = archive.read("openearth_skill_sdk/xskill.py").decode("utf-8") + assert "exactly one" not in source + assert "oracle_multi_atom" in source + + package = types.ModuleType("openearth_skill_sdk") + package.__path__ = [] # type: ignore[attr-defined] + sys.modules["openearth_skill_sdk"] = package + + # Minimal stubs for relative imports used at module import time. + for name in ( + "analyst", + "backend", + "config", + "contracts", + "service", + ): + stub = types.ModuleType(f"openearth_skill_sdk.{name}") + if name == "contracts": + class ScoredAtomInput: # noqa: D401 - test stub + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + stub.ExistingSkillInput = object + stub.ScoredAtomInput = ScoredAtomInput + stub.TrainingResult = object + if name == "config": + stub.load_config = lambda *a, **k: {} + stub.role_config = lambda *a, **k: {} + if name == "analyst": + stub.Analyst = object + if name == "backend": + stub.OpenCodeBackend = object + if name == "service": + stub.TrajectorySkillDistiller = object + sys.modules[f"openearth_skill_sdk.{name}"] = stub + + spec = importlib.util.spec_from_loader( + "openearth_skill_sdk.xskill", + loader=None, + ) + module = importlib.util.module_from_spec(spec) + sys.modules["openearth_skill_sdk.xskill"] = module + exec(compile(source, "xskill.py", "exec"), module.__dict__) + + atom_a = types.SimpleNamespace( + atom_id="a1", + content="first", + ux_score=None, + intent="", + summary="", + used_skills=(), + offset_start=0, + offset_end=1, + ) + atom_b = types.SimpleNamespace( + atom_id="a2", + content="second", + ux_score=None, + intent="", + summary="", + used_skills=(), + offset_start=1, + offset_end=2, + ) + traj = types.SimpleNamespace( + atom_split_status="ready", + atoms=(atom_a, atom_b), + source="temp", + trajectory_id="traj_temp_case", + id="wd1:traj_temp_case", + ) + oracle = { + "schema": 1, + "trajectories": { + "traj_temp_case": {"ux_score": 9, "case_id": "c1"}, + }, + } + scored = list(module._atom_inputs([traj], oracle_scores=oracle)) + assert len(scored) == 2 + assert {item.atom_id for item in scored} == {"a1", "a2"} + assert all(item.ux_score == 9 for item in scored) + assert all(item.score_source == "oracle" for item in scored) diff --git a/examples/kernels/openearth/wheels/openearth_skill_sdk-0.9.0-py3-none-any.whl b/examples/kernels/openearth/wheels/openearth_skill_sdk-0.9.0-py3-none-any.whl new file mode 100644 index 00000000..1b113833 Binary files /dev/null and b/examples/kernels/openearth/wheels/openearth_skill_sdk-0.9.0-py3-none-any.whl differ