refactor(delivery): unify history-to-obligation rules in TypeScript

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md

@@ -2504,6 +2504,12 @@ owns business-rule unification and caller deletion; this RFC owns one durable
 truth, recovery and cutover. Native CLI conversion and a daemon are not
 prerequisites, and PostgreSQL deployment must not hold local adoption hostage.
 
+Delivery history now shares one TS outcome/scale/follow-through read projection
+across status and quota, deleting the replaced Python decisions. This advances
+the TS RFC independently: it changes no provider, durable history, writer fence
+or promotion eligibility. Markdown remains a readable projection; neither it
+nor narrative history labels become an additional delivery authority.
+
 ```text
 CLI / Agent / Dashboard → one TS Todo transaction owner → canonical authority
                                                         ├ structured consumers

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md

@@ -1985,6 +1985,11 @@ Markdown 是**长期保留的一等可读投影**。退役的是它的数据库
 负责业务规则统一和 caller 删除；本 RFC 负责唯一 durable truth、恢复和 cutover。
 CLI 原生 TS 化与 daemon 不是前提，PostgreSQL 部署不能阻塞本地采用。
 
+交付历史现已由 status 与 quota 共用一份 TS outcome/scale/follow-through 读投影，
+删除被替代的 Python decision。这独立推进 TS RFC，不改变 provider、持久历史、
+writer fence 或 promotion 资格。Markdown 仍作为可读投影；它和历史叙述标签都
+不能成为额外的交付权威。
+
 ```text
 CLI / Agent / Dashboard → 唯一 TS Todo 事务 owner → canonical authority
                                                    ├ structured consumers

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -150,12 +150,18 @@ follow-through obligation, prove an outcome, or classify delivery scale.
 For example, `unblocked after dependency update` is not a blocker receipt and
 `implemented network protocol parser` is not preparation-only evidence.
 
-The owning modules remain `control_plane/work_items/delivery_outcome.py`,
-`delivery_signals.py`, and `outcome_followthrough.py`. This is a correctness
-prerequisite inside the existing owner, not a new capability/provider or a
-completed TypeScript transaction migration. It deletes keyword inference and
-its status constants without adding a runtime crossing, schema, or service.
-The existing typed blocker-settlement predicate is reused rather than copied.
+`control_plane/work_items/delivery_history.ts` now owns the complete delivery
+history-to-obligation read projection: outcome, turn kind, scale, consecutive
+streaks and follow-through. Status selects one bounded history batch before one
+`work_item.delivery_history.project` request; quota's latest-run consumer uses
+the same projection with one row. This adds a managed-runtime crossing where
+Python previously decided locally, not one request per field or historical row.
+The Python bridge sends compact typed facts, never narrative or evidence bodies;
+display-only classification is attached after the decision. The replaced
+`delivery_signals.py`, `outcome_followthrough.py`, turn-kind inference and status
+streak wrappers are deleted. Existing TS blocker binding is reused. Python enum
+codecs and the settlement writer predicate still have real callers and remain;
+this is not a writer/transaction or provider migration.
 
 The acceptance invariant is **narrative non-interference**: holding typed
 fields and configuration fixed, rewriting narrative or adding an unvalidated
@@ -182,18 +188,20 @@ label; no legacy prediction is retained without a concrete display consumer.
   erroneous behavior. This intentionally changes status, handoff/review, and
   quota decisions previously derived from untyped historical labels.
 
-Within this delivery domain, the migration unit is the complete
-delivery-history-to-obligation projection, including scale/outcome streaks and
-its status/quota consumers. This defines the slice boundary without displacing
-the provider-first Todo sequence below.
-It must cross at most once per bounded history batch, delete the replaced
-Python decision path, preserve independently reviewed typed cases, and retain
-narrative-mutation regressions through the real CLI. Transport-only golden
-parity is insufficient because the old inference was incorrect. Separately
-inventory writers still omitting material-result fields and retire obsolete
+The migration preserves independently characterized legal typed behavior and
+validates real refresh/history/status/quota entrypoints, batch cardinality and
+narrative non-interference. One intentional correction is separate from parity:
+two invalid work-item identifiers must not compare equal merely because both
+normalize to a missing value. Such observations cannot infer blocker writeback
+or discharge a follow-through obligation. The remaining Python writer predicate
+rejects that case too; no active history is rewritten.
+
+Next, inventory writers still omitting material-result fields and retire obsolete
 marker/hint configuration with an explicit compatibility plan. Exact legacy
-lifecycle classification codes and unrelated cadence policies are outside this
-slice; they must not be reported as migrated or globally free of prose rules.
+lifecycle classification codes, history selection and unrelated cadence policies
+remain outside this slice. Do not claim all writers migrated or all prose rules
+retired. This read-policy closure does not displace the provider-first Todo
+sequence below or wait for a provider cutover.
 
 ### Legacy field-rule retirement checkpoint
 

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -120,11 +120,16 @@ receipt 过期。
 不能判定交付规模。例如，`unblocked after dependency update` 不构成 blocker
 receipt，`implemented network protocol parser` 不构成仅完成准备工作的证据。
 
-规则继续由 `control_plane/work_items/delivery_outcome.py`、`delivery_signals.py`
-和 `outcome_followthrough.py` 持有。本批在既有 owner 中完成正确性前置修复，不新增
-capability/provider，也不宣称完成 TypeScript 事务迁移。删除关键词推断与 status
-常量，不增加 runtime crossing、schema 或 service；复用已有 typed blocker
-settlement 判定，不复制证据绑定规则。
+`control_plane/work_items/delivery_history.ts` 现在持有完整的交付历史到后续义务
+读投影：outcome、turn kind、scale、连续计数及 follow-through。Status 先选出一批
+有界历史，再调用一次 `work_item.delivery_history.project`；quota 的 latest-run
+消费者以单行调用同一投影。相比此前 Python 本地判断，新增 managed-runtime crossing，
+但不是每个字段或每条历史各跨一次。Python bridge 只发送紧凑 typed facts，不发送
+叙述或证据正文；classification 在决策之后作为展示标签附加。
+删除被替代的 `delivery_signals.py`、`outcome_followthrough.py`、turn-kind 推断
+及 status 连续计数 wrapper，复用已有 TS blocker 绑定规则。Python enum codec 和
+settlement writer predicate 仍有真实 caller，因此保留；本批不是 writer/事务或
+provider 迁移。
 
 验收不变量是**叙述非干涉**：固定 typed fields 与配置，改写叙述或增加未经验证的
 `compact_evidence` / `case_result` 对象，都不能改变交付语义与后续执行义务。
@@ -144,13 +149,16 @@ classification 保留为历史标签；没有明确展示消费者时，不保
   词语不再分类 run。不改写持久历史，也不新增开关恢复错误行为。此前由未结构化
   历史标签推导的 status、handoff/review 和 quota 决策会发生明确的行为变化。
 
-交付领域的迁移单元是完整的 delivery-history-to-obligation projection，包含规模／结果
-连续计数与 status/quota 消费者。这定义该领域的切片边界，不改变下文 provider-first
-Todo 的交付顺序。每批有界历史最多跨 runtime 一次，删除被替代的
-Python decision，保留独立审阅的 typed case，并通过真实 CLI 验证叙述变异用例。
-旧推断本身错误，因此只有传输 golden parity 不够。另行盘点仍缺少 material-result
-字段的 writer，并用明确兼容计划退役旧 marker/hint 配置。本批不迁移精确的旧
-lifecycle classification code 或其他 cadence policy，不能宣称全局已无文本规则。
+迁移保留独立刻画的合法 typed 行为，并验证真实 refresh/history/status/quota 入口、
+批次基数与叙述非干涉。有一项有意修正单独披露，不能混称 parity：两个非法 work-item
+identifier 不能仅因都归一化为缺失值而被视为相等；此类 observation 不能推断出
+blocker writeback 或解除后续义务。仍在使用的 Python writer predicate 同样拒绝该
+情况，不改写任何活跃历史。
+
+下一步另行盘点仍缺少 material-result 字段的 writer，并用明确兼容计划退役旧
+marker/hint 配置。精确的旧 lifecycle classification code、历史选取与其他 cadence
+policy 不在本批范围内，不能宣称所有 writer 已迁移或全局已无文本规则。这一读策略
+闭合不改变下文 provider-first Todo 顺序，也不等待 provider cutover。
 
 ### Legacy 字段规则退役检查点
 

examples/control_plane/delivery-signals-readmodel-smoke.py

@@ -1,112 +1,31 @@
 #!/usr/bin/env python3
-from __future__ import annotations
-
+"""The public status batch preserves typed streaks and unknown boundaries."""
 from pathlib import Path
 import sys
 
-
 ROOT = Path(__file__).resolve().parents[2]
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
-
-from loopx import status as status_module  # noqa: E402
-from loopx.control_plane.work_items import delivery_signals as delivery_signal_read_model  # noqa: E402
-
-
-PROFILE = {
-    "outcome_floor": {
-        "outcome_markers": ["merged", "validated"],
-        "surface_only_hints": ["doc_only", "contract"],
-    }
-}
-
-
-def _projection_scale(run: dict[str, object]) -> str:
-    return delivery_signal_read_model.delivery_batch_scale_for_run(run)
-
-
-
-def _projection_outcome(run: dict[str, object], profile: dict[str, object] | None = None) -> str:
-    return delivery_signal_read_model.delivery_outcome_for_run(
-        run,
-        profile,
-        execution_profile_outcome_floor=status_module.execution_profile_outcome_floor,
-    )
-
-
-def _projection_outcome_floor_configured(profile: dict[str, object] | None) -> bool:
-    return delivery_signal_read_model.outcome_floor_configured(
-        profile,
-        execution_profile_outcome_floor=status_module.execution_profile_outcome_floor,
-    )
-
+sys.path.insert(0, str(ROOT))
 
-def _projection_outcome_gap_streak(runs: list[dict[str, object]], profile: dict[str, object] | None) -> int:
-    return delivery_signal_read_model.outcome_gap_streak(
-        runs,
-        profile,
-        delivery_outcome_for_run=_projection_outcome,
-        outcome_floor_configured=_projection_outcome_floor_configured,
-    )
+from loopx.status import project_post_handoff_history  # noqa: E402
 
 
-def _projection_small_scale_streak(runs: list[dict[str, object]]) -> int:
-    return delivery_signal_read_model.small_delivery_batch_scale_streak(
-        runs,
-        delivery_batch_scale_for_run=_projection_scale,
-        small_delivery_batch_scales=status_module.SMALL_DELIVERY_BATCH_SCALES,
-    )
-
-
-def assert_delivery_signal_wrapper_parity() -> None:
-    scale_cases = [
-        {"delivery_batch_scale": "multi_surface", "classification": "ignored_smoke"},
-        {"delivery_batch_scale": "future_scale", "classification": "ignored_smoke"},
-        {"classification": "owner_handoff_consumer_test"},
-        {"classification": "delivery_ranker_readiness_batch"},
-        {"classification": "feedback_reranker_adapter_slice"},
-        {"classification": "single_note"},
-        {},
-    ]
-    for run in scale_cases:
-        assert status_module.delivery_batch_scale_for_run(run) == _projection_scale(run), run
-
-    outcome_cases = [
-        ({"delivery_outcome": "outcome_progress", "classification": "ignored"}, PROFILE),
-        ({"delivery_outcome": "future_outcome", "classification": "ignored"}, PROFILE),
-        ({"classification": "adapter_validated"}, PROFILE),
-        ({"classification": "doc_only_contract"}, PROFILE),
-        ({"classification": "needs_real_outcome"}, PROFILE),
-        ({"classification": "adapter_validated"}, None),
-        ({}, PROFILE),
-    ]
-    for run, profile in outcome_cases:
-        assert status_module.delivery_outcome_for_run(run, profile) == _projection_outcome(run, profile), run
-
-    assert status_module.outcome_floor_configured(PROFILE) == _projection_outcome_floor_configured(PROFILE)
-    assert status_module.outcome_floor_configured(None) == _projection_outcome_floor_configured(None)
-
+def main() -> None:
+    profile = {"outcome_floor": {"outcome_markers": ["validated"]}}
     runs = [
-        {"classification": "needs_real_outcome"},
-        {"classification": "doc_only_contract"},
-        {"classification": "adapter_validated"},
-        {"classification": "needs_real_outcome"},
-    ]
-    assert status_module.outcome_gap_streak(runs, PROFILE) == _projection_outcome_gap_streak(runs, PROFILE)
-    assert status_module.outcome_gap_streak(runs, None) == _projection_outcome_gap_streak(runs, None)
-
-    small_runs = [
-        {"classification": "single_note"},
-        {"classification": "owner_handoff_consumer_test"},
-        {"classification": "delivery_ranker_readiness_batch"},
+        {"delivery_outcome": "surface_only", "delivery_batch_scale": "single_segment"},
+        {"classification": "contract validated implementation"},
+        {"delivery_outcome": "outcome_gap", "delivery_batch_scale": "test_only"},
     ]
-    assert status_module.small_delivery_batch_scale_streak(small_runs) == _projection_small_scale_streak(
-        small_runs
-    )
-
-
-def main() -> None:
-    assert_delivery_signal_wrapper_parity()
+    result = project_post_handoff_history(runs, profile)
+    assert result["post_handoff_small_scale_streak"] == 1
+    assert result["post_handoff_outcome_gap_streak"] == 1
+    assert result["post_handoff_latest_run"]["delivery_batch_scale"] == "single_surface"
+    assert result["post_handoff_recent_runs"][1]["delivery_outcome"] == "unknown"
+    no_floor = project_post_handoff_history([{}])
+    assert "post_handoff_outcome_gap_streak" not in no_floor
+    assert "delivery_outcome" not in no_floor["post_handoff_latest_run"]
+    assert no_floor["post_handoff_latest_run"]["delivery_turn_kind"] == "unknown"
+    print("delivery-signals-readmodel-smoke ok")
 
 
 if __name__ == "__main__":

examples/control_plane/status-markdown-smoke.py

@@ -18,12 +18,11 @@
 if str(REPO_ROOT) not in sys.path:
     sys.path.insert(0, str(REPO_ROOT))
 
+from loopx.status import project_post_handoff_history  # noqa: E402
 from loopx.status import (  # noqa: E402
     build_status_runtime_summaries,
     build_contract_health_projection,
     collect_status,
-    delivery_batch_scale_for_run,
-    delivery_outcome_for_run,
     project_asset_summary_is_public_safe,
 )
 from loopx.presentation.renderers.status_markdown import render_status_markdown  # noqa: E402
@@ -921,8 +920,11 @@ def assert_connected_delivery_surface_loop(payload: dict, markdown: str) -> None
 
 
 def assert_delivery_semantics_require_structured_fields() -> None:
+    def delivery_scale_signal(run: dict) -> str:
+        return project_post_handoff_history([run])["post_handoff_latest_run"]["delivery_batch_scale"]
+
     assert (
-        delivery_batch_scale_for_run(
+        delivery_scale_signal(
             {
                 "classification": "dashboard_home_browser_smoke_regression",
                 "delivery_batch_scale": "multi_surface",
@@ -931,21 +933,21 @@ def assert_delivery_semantics_require_structured_fields() -> None:
         == "multi_surface"
     )
     assert (
-        delivery_batch_scale_for_run(
+        delivery_scale_signal(
             {"classification": "side_bypass_validation_plan_source_shape_consumer_test"}
         )
         == "unknown"
     )
     assert (
-        delivery_batch_scale_for_run({"classification": "owner_handoff_consumer_test"})
+        delivery_scale_signal({"classification": "owner_handoff_consumer_test"})
         == "unknown"
     )
     assert (
-        delivery_batch_scale_for_run({"classification": "delivery_ranker_readiness_batch"})
+        delivery_scale_signal({"classification": "delivery_ranker_readiness_batch"})
         == "unknown"
     )
     assert (
-        delivery_batch_scale_for_run({"classification": "feedback_reranker_adapter_slice"})
+        delivery_scale_signal({"classification": "feedback_reranker_adapter_slice"})
         == "unknown"
     )
     profile = {
@@ -956,31 +958,19 @@ def assert_delivery_semantics_require_structured_fields() -> None:
         }
     }
     assert (
-        delivery_outcome_for_run(
-            {"classification": "side_bypass_owner_drop_landing_forecast_implementation"},
-            profile,
-        )
+        project_post_handoff_history([{"classification": "side_bypass_owner_drop_landing_forecast_implementation"}], profile)["post_handoff_latest_run"]["delivery_outcome"]
         == "unknown"
     )
     assert (
-        delivery_outcome_for_run(
-            {"classification": "side_bypass_ranker_fit_metric_implementation"},
-            profile,
-        )
+        project_post_handoff_history([{"classification": "side_bypass_ranker_fit_metric_implementation"}], profile)["post_handoff_latest_run"]["delivery_outcome"]
         == "unknown"
     )
     assert (
-        delivery_outcome_for_run(
-            {"classification": "side_bypass_macro_evidence_segment_implementation"},
-            profile,
-        )
+        project_post_handoff_history([{"classification": "side_bypass_macro_evidence_segment_implementation"}], profile)["post_handoff_latest_run"]["delivery_outcome"]
         == "unknown"
     )
     assert (
-        delivery_outcome_for_run(
-            {"classification": "status_refresh_without_marker", "delivery_outcome": "primary_goal_outcome"},
-            profile,
-        )
+        project_post_handoff_history([{"classification": "status_refresh_without_marker", "delivery_outcome": "primary_goal_outcome"}], profile)["post_handoff_latest_run"]["delivery_outcome"]
         == "primary_goal_outcome"
     )