[feat-semantic] Add differential semantic dataflow execution

README.md

@@ -101,10 +101,9 @@ examples/helloworld/
 
 ## Design Documents
 
-- `docs/design/v0.0-interface.md`: current v0.0 interface boundary
 - `docs/design/operator-api.md`: DataFrame-style semantic operator API
-- `docs/design/groupby-aggregation-api-rationale.md`: chain aggregation rationale
-- `docs/design/TODO.md`: unresolved v0.0 -> v0.1 interface and research questions
+- `docs/design/lotus-lowering.md`: LOTUS physical lowering and execution contracts
+- `docs/design/TODO.md`: unresolved interface, runtime, and research questions
 - `docs/optimization/incremental-semantic-view-maintenance.tex`: semantic IVM theory draft
 
 ## Development

docs/design/TODO.md

@@ -1,8 +1,52 @@
 # TODO: Semantic IVM Implementation Questions
 
-Status: the current golden line is `docs/optimization/incremental-semantic-view-maintenance.tex` plus  `docs/design/operator-api.md`. This file tracks unresolved implementation questions. It is intentionally question-first, not a decision spec.
-
-## V0.0 -> V0.1 Interface Contracts
+Status: the current golden line is
+`docs/optimization/incremental-semantic-view-maintenance.tex` plus
+`docs/design/operator-api.md`. This is the single current backlog for unresolved
+interface, operator, rule, runtime, and optimizer questions. It is intentionally
+question-first, not a decision spec.
+
+## Current Consolidated Backlog
+
+### Operator execution
+
+- Audit LOTUS Cascade with real traces before exposing it as a supported physical
+  profile. Keep examples, raw outputs, explanations, helper decisions, and stats
+  in a trace side-channel rather than logical result columns.
+- Add an indexed or hybrid execution path for `sem_topk` without changing its
+  logical ranking contract.
+- Evaluate candidate pruning and indexed access for `sem_groupby`; pair batching
+  improves stability but does not reduce the number of semantic pairs by itself.
+- Validate large-group structured `sem_agg` for context limits, retries, trace,
+  and cost before claiming that path is production-stable.
+- Add `min_by` / `arg_min` only when a view needs the payload associated with a
+  minimum ordering value. Removing a current minimum still requires retraction
+  state or recomputation.
+
+### Differential rules and runtime
+
+- Define source deletion and general negative-delta semantics.
+- Define old-state binding or an explicit full-recompute fallback for semantic
+  joins over arbitrary subqueries.
+- Define correct maintenance for materialized left, right, and outer semantic
+  joins, where a new match can retract an old unmatched row.
+- Support dependencies that mix base-log input with changed upstream views.
+- Add recursive/fixpoint planning only with an explicit recursive-view contract.
+- Add physical optimization such as indexed join/aggregate state and shared
+  arrangements without changing logical or maintenance fingerprints silently.
+- Add spec-time placeholder validation and view-aware planner errors once column
+  scope inference is reliable.
+- Define when a standalone semantic aggregate's current output is sufficient
+  maintenance state; schema compatibility alone is not a proof.
+
+### Runtime scheduling
+
+- Define concurrent and multi-writer ordering beyond the current single-process
+  append sequence.
+- Keep maintenance batching, refresh gates, and background consolidation as
+  runtime scheduling decisions unless they change the logical input scope.
+
+## Interface Contracts
 
 - Shared finite dependency maintenance is represented by `DifferentiatedPolicy`
   and executed by `PolicyExecutor`; see
@@ -52,28 +96,6 @@ Status: the current golden line is `docs/optimization/incremental-semantic-view-
   v0.1 should decide whether to hide the expression behind `_expr` or promote a
   stable debug/inspection API.
 
-- The first stateful differential rewrite rule must decide how executable `ΔQ`
-  references current view state `V` and changed input rows `ΔD`. v0.0 keeps
-  `ΔD` as a mathematical/runtime concept rather than modeling it as a
-  `QueryExpr` operator; current row-local fragment rules work by binding the
-  source log to changed rows at runtime.
-
-- The first `sem_groupby(...).sem_agg(...)` view-boundary rule is implemented as
-  a Claude-style full-next-view query shape:
-
-  ```text
-  ΔC = ΔD.sem_flat_map(...)
-  ΔT = ΔC.sem_groupby(...).sem_agg(...)
-  M = ΔT.sem_join(V, how="outer", instruction=<same topic identity instruction>)
-  V' = M.sem_map(instruction=<same consolidation instruction>)
-  ```
-
-  Current limits: it only applies at a public view boundary, returns full `V'`
-  rather than minimal `DeltaV`, and is not a generic exact `sem_agg`
-  maintenance proof. Deterministic placeholder rewrite is separate from future
-  semantic prompt rewriting. If a later design computes `DeltaV` instead, the
-  same logical maintenance may become `left join + upsert/delete/skip`.
-
 - Adapter capability and default adapter injection are still unresolved. Before
   adding non-LOTUS engines, decide what capability contract each execution
   adapter exposes and where the default adapter is injected.
@@ -83,28 +105,6 @@ Status: the current golden line is `docs/optimization/incremental-semantic-view-
   on the adapter dataclass or are delegated to LOTUS/provider-native
   configuration.
 
-
-## Claude Code native memory path alignment
-
-Detailed native Claude Code memory analysis now lives in
-`docs/design/native-claude-memory-audit.zh.md`. Keep that document as the
-canonical design note instead of duplicating long native-path analysis here.
-
-Current TODOs:
-
-- Keep native comparison experiments separated by path: main-agent inline write,
-  `extractMemories` background extraction, and `autoDream` consolidation.
-- Do not use a CLI `-p` LOCOMO replay as evidence that `extractMemories` or
-  `autoDream` ran; it is only a native CLI / main-agent behavior audit unless
-  debug artifacts prove otherwise.
-- Add an extract-only component harness before claiming parity with
-  `agent-memory` `sem_flat_map -> sem_groupby.sem_agg`.
-- Treat `--turn-size 1` native CLI runs as one-row-per-turn behavior, not strict
-  one-message extractor tests.
-- If comparing consolidation, prefer an extract-only + seeded `autoDream` gate
-  experiment so native `autoDream` itself runs while the scheduler conditions are
-  explicit.
-
 ## Open Research Questions
 
 ### API context fields

docs/design/groupby_agg.md

@@ -7,7 +7,8 @@
   `input_cols` 细节。
 - 第三部分记录 implementation-wise differential rules，说明实际 lowering 里
   `sem_agg(input_cols=...)`、`collect_list(...)`、`flatten(...)`、`min(...)`
-  和 `least(...)` 带来的 rule 变体。
+  和 `least(...)` 带来的 rule 变体，并单独说明普通 relational
+  `count/sum/avg` 的精确增量状态。
 
 这里的 rules 分成 paper-wise 和 implementation-wise 两层。本文档不承诺当前
 implementation 已经支持所有 rules，也不决定 optimizer 如何选择 rule。
@@ -114,6 +115,44 @@ input 输出零组。
 keys 重名；重名会直接报错。作为 mixed `group_by(keys).agg(...)` 中的 aggregate
 spec 使用时，则遵循该 mixed `.agg(...)` 的 key precedence contract。
 
+#### `count`、`sum` 和 `avg`
+
+这三个是普通 relational aggregates，不调用 LLM：
+
+```python
+overall = rows.agg(
+    count(output_col="row_count"),
+    sum(column="amount", output_col="total_amount"),
+    avg(column="amount", output_col="average_amount"),
+)
+
+by_region = rows.group_by("region").agg(
+    count(output_col="row_count"),
+    sum(column="amount", output_col="total_amount"),
+    avg(column="amount", output_col="average_amount"),
+)
+```
+
+它们的合同是：
+
+```text
+Relation.agg(...):
+  many rows -> one output row
+  output schema = aggregate output columns
+
+group_by(K).agg(...):
+  many rows per exact key -> one output row per key
+  output schema = K + aggregate output columns
+```
+
+`count` 是 `COUNT(*)`，包括含 null 的 row。`sum` 和 `avg` 忽略 null；没有非 null
+数值时输出 null。空的 global relation 仍输出一行，其中 count 为 0，sum/avg 为
+null；空的 grouped relation 输出零行。
+
+当前 numeric aggregate family 只能包含 `count/sum/avg`，不能在同一次 `.agg(...)`
+里混入 `sem_agg`、`array_agg`、`collect_list` 或 `min`。它只适用于普通 `Relation`
+和 deterministic `group_by`；`sem_groupby(...).count/sum/avg(...)` 尚未实现。
+
 ### 1.3 `sem_groupby(keys).sem_agg(...)`
 
 这里的 `keys` 指 `sem_groupby(input_cols=keys, ...)` 里的 semantic keys。它们是
@@ -261,6 +300,12 @@ Rules:
 - `am.min(column=..., output_col=...)` 或
   `am.min(columns=[...], output_col=...)` 可以和 `sem_agg`、`array_agg`、
   `collect_list` 一起作为 aggregate spec。
+- 普通 `Relation` 或 deterministic `group_by` 可以使用只包含
+  `count/sum/avg` 的 numeric aggregate family。
+- Numeric aggregate family 暂时不能和上面的 semantic/evidence aggregate
+  family 混用，也不能用于 `sem_groupby`。
+- Numeric aggregate outputs 不能和 deterministic keys 重名；多个 outputs
+  之间也不能重名。
 
 例子：
 
@@ -489,12 +534,42 @@ JoinGroups(GΔ, V, Kr, θg, M) =
 
 `group_match(θg)`是compiler从原`sem_groupby` instruction改写出的“changed group
 和current group是否属于同一组”谓词。`Ks`只是这个semantic predicate读取的内容，
-绝不能写成`on=Ks`：`on`只接受必须精确相等的`partition_by` keys `Kr`。
+绝不能写成`on=Ks`。`sem_join.on`本身支持general deterministic `JoinOn`；但在这条
+grouped-aggregate rule里，`on=Kr`有一个更窄、更明确的职责：只用必须精确相等的
+`partition_by` keys排除跨partition candidates。
 
 `MergeByTarget`表示：matched rows按选中的current target合并；没有target的changed
 group形成新row；未被触及的current row原样保留。如果多个changed groups命中同一个
 target，只生成一次合并后的target row。
 
+### 可复用的 direct `sem_join` 增量规则
+
+`rule-join-map`会在自己的maintenance query内部调用`sem_join`，但direct
+`sem_join` view的增量维护是一个独立、可复用的binary-state能力。对：
+
+```text
+J = L.sem_join(R, on=φ, instruction=θ, how="inner", k=None)
+```
+
+append-only rule是：
+
+```text
+ΔJ = SemJoin(ΔL, R0, φ, θ)
+     bag-union SemJoin(L0, ΔR, φ, θ)
+     bag-union SemJoin(ΔL, ΔR, φ, θ)
+
+J1 = J0 bag-union ΔJ
+```
+
+这里`φ`可以是`on=None`、同名exact keys或general deterministic `JoinOn`。每一项都
+先用`φ`缩小candidate pairs，再执行原semantic predicate `θ`；`ΔL × ΔR`只在第三项
+计算一次。实现使用保留重复行的`concat`，不是会去重的`union`。
+
+当前direct differential只支持append-only、`how="inner"`和`k=None`。这不限制
+静态`sem_join`，也不改变本章的grouped-aggregate rules：`rule-join-map`仍然可以在
+其自身state-merge lowering内部使用outer或top-k semantic join；它不是把该内部
+join再次编译成一个独立的direct differential node。
+
 ### 2.1 `sem_groupby(...).sem_agg(...)`
 
 ```text
@@ -661,6 +736,38 @@ semantic keys `Ks`仍然只参与`θg`，不要求精确相等。
 `Kr`找到受影响的partition；只有这些partition内部的changed semantic groups需要继续
 执行join-map或re-group。未被delta触及的partition原样保留。
 
+### 2.9 Exact relational `count/sum/avg`
+
+普通 relational aggregate 不需要 semantic rule family。它把每个 exact key 的
+输入压成可加减的状态：
+
+```text
+S(D) = (
+  row_count(D),
+  numeric_sum(D),
+  non_null_count(D)
+)
+```
+
+设 `Δ+` 是 parent 插入，`Δ-` 是 parent 撤回：
+
+```text
+S' = S + S(Δ+) - S(Δ-)
+
+count = row_count
+sum   = null if non_null_count = 0 else numeric_sum
+avg   = null if non_null_count = 0 else numeric_sum / non_null_count
+```
+
+有 `group_by(K)` 时，同一公式按 `K` 分开执行：
+
+```text
+S'[k] = S[k] + S(Δ+[k]) - S(Δ-[k])
+```
+
+说人话就是：新row属于哪个普通key，就只更新那个key的计数和总和；旧row被上游确定性
+变化撤回时，再从同一个key减掉。旧key之间不做join，也不重新扫描未变化的groups。
+
 ## 3. Implementation-wise differential rules
 
 paper-wise rules 里不写 `input_cols`。implementation 里必须写，因为
@@ -900,18 +1007,43 @@ rule-join-map:
 semantic outer join，再在结果上filter Kr”，因为那会丢失本应保留的unmatched rows，
 改变outer-join语义。
 
+### 3.10 Exact relational aggregate state
+
+实现不会把 `avg` 当成一个可以直接相加的输出值。Planner为整组
+`count/sum/avg`建立一个隐藏的 aggregate-state node：
+
+```text
+logical query:
+  D[.group_by(K)].agg(count, sum, avg)
+
+hidden state:
+  K, row_count, numeric_sum, non_null_count
+
+public finalizer:
+  K, declared count/sum/avg outputs
+```
+
+同一个 input column 被 `sum` 和 `avg` 同时使用时，只维护一份 sum 和 non-null
+count。Runtime把 parent的`inserted_rows`加进state，把`retracted_rows`减出state；
+grouped state的row count变成0时删除该group，global state则始终保留一行。隐藏列会进入
+checkpoint，但不会进入public view schema。
+
+这里没有`rule-join-map`、`rule-re-group`或`rule-all-group`选择，因为key和状态更新都是
+确定性的。它也不改变`min`、`array_agg`或semantic aggregates现有的retraction边界。
+
 ## 4. 本文档不定义什么
 
 本文档不决定：
 
 - optimizer 应该选择哪个 rule family；
-- delete / negative delta 时保存 raw group、ordered state，还是 full recompute；
+- `min`、`array_agg` 和 semantic aggregates 遇到 delete / negative delta 时保存
+  raw group、ordered state，还是 full recompute；
 - DAG runtime 应该使用哪种 indexed physical aggregate state。
 
 当前 implementation 已用真实 `QueryExpr` operator graph 实现本文第三章覆盖的
-`sem_agg`、`array_agg`、mixed `.agg(...)` 和 `min` lowering；不存在 public 或
-internal `agg_merge` operator。Shared `PolicyExecutor` 会在这些 stateful node
-boundary 对 insert-only change 执行本章定义的 maintenance query。如果 parent
-change 含 internal retraction，executor 会在 parent 的完整 next state 上重算当前
-semantic aggregate node，再把 replacement change 传给下游。Source delete 和更
-高效的 indexed aggregate state 仍未实现。
+`sem_agg`、`array_agg`、mixed `.agg(...)`、`min` 和 numeric `count/sum/avg`
+lowering；不存在 public `aggregate_state` 或 `agg_merge` operator。Numeric
+aggregates直接维护hidden additive state，可以消费parent产生的insertions和
+retractions。现有semantic aggregate node遇到parent retraction时，仍会在parent完整
+next state上重算，再把replacement change传给下游。Source delete API、numeric与
+semantic aggregate混合，以及`sem_groupby`后的numeric aggregates仍未实现。