fix: validate realtime report partitions by scope

.agents/skills/odps-product-efficiency-report/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: odps-product-efficiency-report
-description: 生成、执行、校验并分析参数化 ODPS 产品效率实验报表，包含 DAU、曝光、播放、分享、回流、STR 和 ROV，并按头部、推荐、全部流量拆分；每次成功查询必须给出数据结论，用户要求飞书时将结论随表格发送。用于产品类型、离线或实时模式、日期范围、rootSessionId 分桶位置、实验/对照尾号、版本过滤或企微排除条件发生变化的查询。
+description: 生成、执行、校验并分析参数化 ODPS 产品效率报表，支持实验分桶对比和不分桶的全量汇总，包含 DAU、曝光、播放、分享、回流、STR 和 ROV，并按头部、推荐、全部流量拆分；每次成功查询必须给出数据结论，用户要求飞书时将结论随表格发送。用于产品类型、离线或实时模式、日期范围、rootSessionId 分桶位置、实验/对照尾号、版本过滤、企微排除条件或仅全量汇总发生变化的查询。
 ---
 
 # ODPS 产品效率报表
@@ -18,10 +18,10 @@ description: 生成、执行、校验并分析参数化 ODPS 产品效率实验
 - 日期或日期范围；“今天”“昨天”等相对日期需解析为具体 `yyyyMMdd` 并回显。
 - 离线或实时数据。
 - 产品类型 `app_type`。
-- `rootSessionId` 倒数第几位分桶。
-- 哪些尾号属于实验组。
 - 全部版本或指定版本。
 
+实验对比还必须确认 `rootSessionId` 倒数第几位分桶及实验组尾号。用户明确要求“仅全量”“不分实验组”时进入全量汇总模式，不再追问分桶位置、实验桶或对照桶。
+
 任一项缺失时，先用一个简短问题集中询问缺失项；不得从旧报表、目录名、历史习惯或默认值猜测。用户已经明确全部口径时无需重复确认。对照组未指定时，只有在实验组已明确后才可使用 `0-f` 的补集，并在执行前回显。
 
 ## 工作流程
@@ -29,22 +29,24 @@ description: 生成、执行、校验并分析参数化 ODPS 产品效率实验
 1. 阅读 [metrics.md](references/metrics.md) 和 [raw-output-contract.md](references/raw-output-contract.md)。
 2. 执行“执行前强制确认”；存在缺失项时停止，不生成文件、不查询、不上传。
 3. 在用户指定输出目录把参数写入 `request.json`。
-4. 规范化并校验参数：
+4. 实验对比模式规范化并校验参数：
 
    `python3 "$SKILL_DIR/scripts/normalize_request.py" request.json > normalized.json`
 
-5. 明确回显日期、模式、数据表、分桶位置和分组、版本策略及企微处理方式。
+   全量汇总模式不运行该分桶参数规范化脚本。
+5. 明确回显日期、模式、数据表、报表范围、版本策略及企微处理方式；实验对比模式还需回显分桶位置和分组。
 6. 生成一个符合原始事实契约的参数化 SQL。不得搜索或依赖外部 `AGENTS.md`、特定日期 SQL 或本地 runner。
-7. 预检全部日期、数据表、产品类型、分桶表达式、版本条件、分组标签和渠道排除条件。离线和实时的活跃、视频日志都必须从 `extparams.$.rootSessionId` 分桶，分享日志使用物理列 `rootsessionid`；特别禁止用离线 `useractive_log.rootsessionid` 计算 DAU。离线 `video_action_log_applet` 和实时 `video_action_log_flow` 都只用各自日期分区、`apptype` 和 `businesstype` 过滤，SQL 中禁止引用物理字段 `business`。分享 PV/UV 使用视频日志的 `businesstype='videoShareFriend'`；回流源分享必须使用分享日志的 `topic='share'`，回流点击必须使用 `topic='click'`，禁止用 `type='share'` 判断源分享。SQL 必须为每天生成完整 16 个桶。
+7. 预检全部日期、数据表、产品类型、版本条件和渠道排除条件。实验对比模式还需校验分桶表达式和分组标签，并为每天生成完整 16 个桶；活跃、视频日志从 `extparams.$.rootSessionId` 分桶，分享日志使用物理列 `rootsessionid`。全量汇总模式不得添加 rootSessionId 分桶。实时 `useractive_log_per5min` 和 `user_share_log_per5min` 必须使用 `dt LIKE 'yyyyMMdd%'` 查询当天累计数据，不能使用 `dt='yyyyMMdd'`。离线 `video_action_log_applet` 和实时 `video_action_log_flow` 都只用各自日期分区、`apptype` 和 `businesstype` 过滤，SQL 中禁止引用物理字段 `business`。分享 PV/UV 使用视频日志的 `businesstype='videoShareFriend'`；回流源分享必须使用分享日志的 `topic='share'`，回流点击必须使用 `topic='click'`，禁止用 `type='share'` 判断源分享。
 8. 只提交一个 ODPS 实例：
 
    `python3 "$SKILL_DIR/scripts/run_sql.py" query.sql raw_facts.csv`
 
-9. 生成标准完整报表和聚合报表：
+9. 实验对比模式生成标准完整报表和聚合报表：
 
    `python3 "$SKILL_DIR/scripts/format_report.py" normalized.json raw_facts.csv full_report.csv --aggregate-output aggregate_report.csv`
 
-10. 执行 [metrics.md](references/metrics.md) 中的全部校验，并按“分析与飞书概要契约”输出结论；若某日 16 桶 DAU 全为零而任一行为事实非零，按 rootSessionId 映射错误停止，不得格式化或分析该结果。不得只返回文件链接或原始指标。
+   全量汇总模式直接发布 SQL 的逐日汇总结果，不调用 16 桶格式器。
+10. 执行 [metrics.md](references/metrics.md) 中的适用校验，并按“分析与飞书概要契约”输出结论；实验对比模式若某日 16 桶 DAU 全为零而任一行为事实非零，按 rootSessionId 映射错误停止，不得格式化或分析该结果。不得只返回文件链接或原始指标。
 11. 仅当用户明确要求上传飞书时，先生成 `summary_message`，再使用 `$feishu-data-publisher` 发布最终 CSV 或 Excel，并通过 `--message` 把该概要写入群卡片；未明确要求时只保留本地结果。
 12. 发布成功后，在对话回复中只给出飞书链接、权限和发送状态，并说明概要已随群卡片发送；不要重复整段结论。
 
@@ -52,8 +54,8 @@ description: 生成、执行、校验并分析参数化 ODPS 产品效率实验
 
 每次成功查询都必须以校验通过的聚合报表为依据生成分析：
 
-- 先回显日期、实时或离线、产品、实验/对照桶、版本和数据截止时间。
-- DAU 使用真实桶数比较每桶均值，同时给出实验组总量；不得直接比较不同桶数的 DAU 总量。
+- 先回显日期、实时或离线、产品、报表范围、版本和数据截止时间；实验对比模式还需回显实验/对照桶。
+- 实验对比模式的 DAU 使用真实桶数比较每桶均值，同时给出实验组总量；不得直接比较不同桶数的 DAU 总量。
 - 分别分析头部、推荐和全部流量，至少覆盖曝光 PV/DAU、播放 PV/DAU、分享 PV/DAU、STR、回流 UV/DAU 和 ROV 的实验值、对照值及相对变化。
 - 明确指出最强正向指标、主要负向或退化指标，以及整体判断。只描述相关变化，不把实验对比表述为已证明的因果关系。
 - 对照组为单桶、实时数据未收全、对照为零或样本过小时必须提示风险；基线为零时写“不可比较”，不得生成无穷变化率。
@@ -62,7 +64,8 @@ description: 生成、执行、校验并分析参数化 ODPS 产品效率实验
 
 ## 参数规则
 
-- 必须提供 `app_type`、`date_from`、`date_to`、`data_mode`、`bucket_position_from_end`、`experiment_buckets` 和 `version`。
+- 两种模式都必须提供 `app_type`、`date_from`、`date_to`、`data_mode` 和 `version`。
+- 实验对比模式还必须提供 `bucket_position_from_end` 和 `experiment_buckets`；全量汇总模式中两者均为 null。
 - 未明确给出对照桶时，自动使用 `0-f` 中实验桶的补集。
 - `version: "all"` 表示不限制版本；指定版本时限制 DAU、视频行为和源分享，不限制回流接收用户。
 - 实时模式只支持一个自然日；已完成的历史日期使用离线数据源。

.agents/skills/odps-product-efficiency-report/agents/openai.yaml

@@ -1,4 +1,4 @@
 interface:
-  display_name: "产品效率实验报表"
-  short_description: "生成并分析产品效率实验报表，按需将结论和表格发送飞书"
+  display_name: "产品效率报表"
+  short_description: "生成并分析产品效率实验对比或全量汇总，按需发送飞书"
   default_prompt: "使用 $odps-product-efficiency-report 查询并分析指定产品效率数据，按需将结论和报表发送飞书。"

.agents/skills/odps-product-efficiency-report/references/raw-output-contract.md

@@ -1,5 +1,13 @@
 # Product-efficiency raw output contract
 
+## Overall-only variant
+
+When the user explicitly requests only the overall total without experiment buckets, emit one row per `stat_date` and do not derive or filter `rootSessionId`. Include the identity columns `stat_date`, `app_type`, and `version_code`, `dau`, all 21 source fact columns below, and for each `head`, `recommend`, and `all` source include exposure/play/share PV per DAU, return UV per DAU, STR, and ROV. Do not create a bucket spine and do not run the 16-bucket formatter for this variant.
+
+For realtime overall totals, `useractive_log_per5min` and both source/click reads of `user_share_log_per5min` must use `dt LIKE 'yyyyMMdd%'`; `video_action_log_flow` uses consistent `year`, `month`, and day-valued `dt` and omits `hh` for current-day cumulative data.
+
+## Experiment-bucket variant
+
 Generate one row per `stat_date + bucket`. Every date must contain exactly the 16 lowercase buckets `0-f`; create a bucket spine and left join zero facts when necessary.
 
 ## Identity columns

findings.md

@@ -49,3 +49,5 @@
 - The same no-`business` rule applies to realtime `video_action_log_flow`: use `year/month/dt`, `apptype`, and `businesstype`, with `rootSessionId` from `extparams`. The repository-local Skill and host contract now enforce this symmetrically for offline and realtime product-efficiency queries.
 - The follow-up “分析下 有什么结论吗” was correctly answered by Codex from the current thread, but the structured decision only allowed `sql` or `skill_script`; the host discarded the answer and raised “Agent 返回 ready 但没有 SQL”. A first-class `direct_reply` execution mode fixes the orchestration mismatch without changing SQL generation or validation.
 - A live post-change routing check confirms the boundary: an ROV definition returned a complete direct answer with no SQL, while a fully specified fresh product-efficiency request still selected `odps-product-efficiency-report`, produced SQL, extracted all parameters, and passed the existing host contract guard.
+- Realtime run `20260812_163712_f891b997` never reached ODPS: all three source CTEs had partition predicates, but the host accepted unqualified partition columns only when the entire SQL contained one partitioned table, so it falsely rejected unambiguous `dt/year/month/hh` conditions inside single-source CTEs.
+- The same saved SQL also used `dt='20260812'` on `useractive_log_per5min` and `user_share_log_per5min`; the verified realtime contract requires `dt LIKE '20260812%'` for current-day cumulative data. Total-only product efficiency also needs a direct aggregate workbook path because the deterministic formatter requires experiment buckets and 16 raw bucket rows.

progress.md

@@ -65,3 +65,8 @@
 - Added `direct_reply` to the structured Codex decision and host dispatcher. Ordinary conversation, explanations, and analysis of existing thread results now reply directly; fresh facts still require the original Skill/SQL or deterministic script route.
 - Added model and service regressions; all 42 tests and Python compilation pass. A real concurrent Codex check returned `direct_reply` for an ROV explanation and a 5,825-character guarded SQL plan for a fully specified offline product-efficiency query; the SQL passed the host product-efficiency contract.
 - Waited for run `20260812_161030_b31d848a` to finish and publish before restarting. Service PID `64474` is running with Feishu WebSocket established and no active query runs; persisted per-user thread context remains intact.
+- Started Phase 17 after realtime run `20260812_163712_f891b997` was falsely rejected for missing partitions before ODPS submission.
+- Reworked partition preflight to validate each physical table occurrence in its own SELECT/CTE scope, accepting unambiguous unqualified fields while still requiring every repeated source occurrence to be filtered.
+- Added realtime product-efficiency validation for `dt LIKE 'yyyyMMdd%'` on per5min active/share sources and a direct workbook path for explicit total-only reports without experiment buckets; updated the repository Skill contract accordingly.
+- A fresh Codex forward check generated a 5,835-character product-0 realtime total-only SQL with null bucket parameters and passed live ODPS metadata preflight without executing a query.
+- All 48 tests, Python compilation, whitespace checks, and Skill quick validation pass. Restarted the Feishu service as PID `72780`; bot identity lookup and WebSocket startup succeeded with no active queries.

src/data_query_agent/service.py

@@ -212,7 +212,11 @@ class DataQueryService:
             })
             try:
                 if current.selected_skill == "odps-product-efficiency-report":
-                    self.guard.validate_product_efficiency_contract(sql, current.parameters.data_mode)
+                    self.guard.validate_product_efficiency_contract(
+                        sql,
+                        current.parameters.data_mode,
+                        bucketed=current.parameters.bucket_position_from_end is not None,
+                    )
                 await validate_for_odps(sql, self.guard, self.odps)
                 self.state.update_run(run_id, "running")
                 result = await self.odps.execute(sql)

src/data_query_agent/skill_executor.py

@@ -127,6 +127,17 @@ class SkillExecutor:
         sql: str,
         info: dict[str, object],
     ) -> SkillArtifact:
+        if skill_name == "odps-product-efficiency-report":
+            bucket_fields = (
+                parameters.bucket_position_from_end,
+                parameters.experiment_buckets,
+            )
+            if bucket_fields == (None, None):
+                _, xlsx_path = write_result_files(run_dir, raw_frame, sql, info)
+                return SkillArtifact(raw_frame, xlsx_path, str(info.get("ODPS instance_id") or ""))
+            if any(value is None for value in bucket_fields):
+                raise ValueError("产品效率分桶报表必须同时提供分桶位置和实验桶")
+
         if skill_name not in REPORT_SKILLS:
             _, xlsx_path = write_result_files(run_dir, raw_frame, sql, info)
             return SkillArtifact(raw_frame, xlsx_path, str(info.get("ODPS instance_id") or ""))

src/data_query_agent/sql_guard.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
 
 from sqlglot import exp, parse, parse_one
@@ -62,11 +63,13 @@ class SQLGuard:
         return refs
 
     @staticmethod
-    def validate_product_efficiency_contract(sql: str, data_mode: str | None) -> None:
+    def validate_product_efficiency_contract(
+        sql: str, data_mode: str | None, *, bucketed: bool = True
+    ) -> None:
         if data_mode not in {"offline", "realtime"}:
             return
         statement = parse_one(sql, read="hive")
-        if data_mode == "offline":
+        if data_mode == "offline" and bucketed:
             active_selects = [
                 table.find_ancestor(exp.Select)
                 for table in statement.find_all(exp.Table)
@@ -88,6 +91,23 @@ class SQLGuard:
                     "禁止使用 useractive_log.rootsessionid"
                 )
 
+        if data_mode == "realtime":
+            invalid_sources: list[str] = []
+            for table in statement.find_all(exp.Table):
+                if table.name.lower() not in {
+                    "useractive_log_per5min",
+                    "user_share_log_per5min",
+                }:
+                    continue
+                select = table.find_ancestor(exp.Select)
+                if select is None or not SQLGuard._has_day_prefix_filter(select, table):
+                    invalid_sources.append(table.name)
+            if invalid_sources:
+                names = "、".join(sorted(set(invalid_sources)))
+                raise SQLValidationError(
+                    f"实时表 {names} 必须使用 dt LIKE 'yyyyMMdd%' 查询当天累计数据"
+                )
+
         for table in statement.find_all(exp.Table):
             if table.name.lower() not in {"video_action_log_applet", "video_action_log_flow"}:
                 continue
@@ -108,37 +128,77 @@ class SQLGuard:
     @staticmethod
     def validate_partition_predicates(sql: str, partitions: dict[str, list[str]]) -> None:
         statement = parse_one(sql, read="hive")
-        predicate_roots: list[exp.Expression] = []
-        predicate_roots.extend(where.this for where in statement.find_all(exp.Where))
-        predicate_roots.extend(
-            on for join in statement.find_all(exp.Join) if (on := join.args.get("on")) is not None
-        )
-        partitioned_count = sum(bool(columns) for columns in partitions.values())
         missing: list[str] = []
         for table, columns in partitions.items():
             if not columns:
                 continue
             table_name = table.rsplit(".", 1)[-1].lower()
-            qualifiers = {
-                candidate.alias_or_name.lower()
+            occurrences = [
+                candidate
                 for candidate in statement.find_all(exp.Table)
                 if candidate.name.lower() == table_name
-            }
+            ]
             partition_names = {column.lower() for column in columns}
             if table_name == "video_action_log_applet":
                 partition_names.discard("business")
-            found = False
-            for predicate in predicate_roots:
-                for candidate in predicate.find_all(exp.Column):
-                    qualifier = candidate.table.lower() if candidate.table else ""
-                    if candidate.name.lower() not in partition_names:
-                        continue
-                    if qualifier in qualifiers or (not qualifier and partitioned_count == 1):
-                        found = True
-                        break
-                if found:
-                    break
-            if not found:
+            if any(
+                not SQLGuard._table_has_partition_filter(occurrence, partition_names)
+                for occurrence in occurrences
+            ):
                 missing.append(f"{table}({', '.join(sorted(partition_names))})")
         if missing:
             raise SQLValidationError("分区表缺少明确分区条件：" + "；".join(missing))
+
+    @staticmethod
+    def _table_has_partition_filter(table: exp.Table, partition_names: set[str]) -> bool:
+        select = table.find_ancestor(exp.Select)
+        if select is None:
+            return False
+        sources = [
+            candidate
+            for candidate in select.find_all(exp.Table)
+            if candidate.find_ancestor(exp.Select) is select
+        ]
+        allow_unqualified = len(sources) == 1
+        qualifier = table.alias_or_name.lower()
+        predicates: list[exp.Expression] = [
+            where.this
+            for where in select.find_all(exp.Where)
+            if where.find_ancestor(exp.Select) is select
+        ]
+        predicates.extend(
+            join.args["on"]
+            for join in select.find_all(exp.Join)
+            if join.find_ancestor(exp.Select) is select and join.args.get("on") is not None
+        )
+        for predicate in predicates:
+            for column in predicate.find_all(exp.Column):
+                if column.name.lower() not in partition_names:
+                    continue
+                column_qualifier = column.table.lower() if column.table else ""
+                if column_qualifier == qualifier or (not column_qualifier and allow_unqualified):
+                    return True
+        return False
+
+    @staticmethod
+    def _has_day_prefix_filter(select: exp.Select, table: exp.Table) -> bool:
+        sources = [
+            candidate
+            for candidate in select.find_all(exp.Table)
+            if candidate.find_ancestor(exp.Select) is select
+        ]
+        allow_unqualified = len(sources) == 1
+        qualifier = table.alias_or_name.lower()
+        for like in select.find_all(exp.Like):
+            if like.find_ancestor(exp.Select) is not select:
+                continue
+            column = like.this
+            pattern = like.expression
+            if not isinstance(column, exp.Column) or column.name.lower() != "dt":
+                continue
+            column_qualifier = column.table.lower() if column.table else ""
+            if column_qualifier != qualifier and not (not column_qualifier and allow_unqualified):
+                continue
+            if isinstance(pattern, exp.Literal) and re.fullmatch(r"\d{8}%", str(pattern.this)):
+                return True
+        return False

task_plan.md

@@ -110,6 +110,13 @@ Build a Python Codex SDK service that receives allowlisted Feishu messages, keep
 - [x] Add regressions, run a real Codex routing check, and restart the Feishu service
 - **Status:** completed
 
+### Phase 17: Fix realtime product-efficiency preflight
+- [x] Reproduce the false missing-partition error from the saved failed run
+- [x] Validate partition predicates within each CTE and every repeated table occurrence
+- [x] Enforce realtime per5min day-prefix filters and support explicit total-only reports
+- [x] Run full regressions, restart the Feishu service, and push the fix
+- **Status:** completed
+
 ## Decisions
 
 - Python 3.12, `openai-codex==0.144.4`, OpenRouter, GPT-5.6 Terra.

tests/test_product_efficiency_skill_contract.py

@@ -28,3 +28,15 @@ def test_generic_catalog_uses_topic_for_return_chain() -> None:
     assert "源分享使用 `topic='share'`" in catalog
     assert "点击使用 `topic='click'`" in catalog
     assert "禁止用 `type='share'` 判断源分享" in catalog
+
+
+def test_realtime_total_only_contract_is_explicit() -> None:
+    skill = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
+    raw_contract = (SKILL_DIR / "references" / "raw-output-contract.md").read_text(
+        encoding="utf-8"
+    )
+
+    assert "仅全量" in skill
+    assert "`dt LIKE 'yyyyMMdd%'`" in skill
+    assert "Overall-only variant" in raw_contract
+    assert "do not run the 16-bucket formatter" in raw_contract

tests/test_skill_executor.py

@@ -98,3 +98,34 @@ async def test_generic_sql_result_uses_existing_workbook_path(tmp_path) -> None:
 
     assert artifact.xlsx_path.is_file()
     assert artifact.instance_id == "i-generic"
+
+
+@pytest.mark.asyncio
+async def test_total_only_product_efficiency_uses_direct_workbook(tmp_path, monkeypatch) -> None:
+    executor = SkillExecutor(replace(Settings.load(), runtime_dir=tmp_path))
+
+    async def unexpected_run(*args, **kwargs):
+        raise AssertionError("total-only report must not invoke the bucket formatter")
+
+    monkeypatch.setattr(executor, "_run", unexpected_run)
+    frame = pd.DataFrame({"stat_date": ["20260812"], "dau": [100], "all_exposure_pv": [200]})
+    artifact = await executor.format_report(
+        "odps-product-efficiency-report",
+        parameters(
+            app_type="0",
+            date_from="20260812",
+            date_to="20260812",
+            data_mode="realtime",
+            bucket_position_from_end=None,
+            experiment_buckets=None,
+            version="all",
+        ),
+        tmp_path,
+        frame,
+        "SELECT 1",
+        {"ODPS instance_id": "i-total"},
+    )
+
+    assert artifact.dataframe.equals(frame)
+    assert artifact.xlsx_path.is_file()
+    assert artifact.instance_id == "i-total"

tests/test_sql_guard.py

@@ -52,6 +52,47 @@ def test_each_partitioned_join_source_must_be_filtered() -> None:
     )
 
 
+def test_unqualified_partition_columns_are_valid_inside_single_source_ctes() -> None:
+    sql = """
+    WITH dau AS (
+      SELECT COUNT(*) FROM loghubods.useractive_log_per5min
+      WHERE dt LIKE '20260812%'
+    ), video AS (
+      SELECT COUNT(*) FROM loghubods.video_action_log_flow
+      WHERE year='2026' AND month='08' AND dt='20260812' AND hh='16'
+    ), shares AS (
+      SELECT COUNT(*) FROM loghubods.user_share_log_per5min
+      WHERE dt LIKE '20260812%'
+    )
+    SELECT * FROM dau CROSS JOIN video CROSS JOIN shares
+    """
+    SQLGuard.validate_partition_predicates(
+        sql,
+        {
+            "loghubods.useractive_log_per5min": ["dt"],
+            "loghubods.video_action_log_flow": ["year", "month", "dt", "hh"],
+            "loghubods.user_share_log_per5min": ["dt"],
+        },
+    )
+
+
+def test_every_repeated_partitioned_source_must_have_its_own_filter() -> None:
+    sql = """
+    WITH source AS (
+      SELECT shareid FROM loghubods.user_share_log_per5min
+      WHERE dt LIKE '20260812%' AND topic='share'
+    ), click AS (
+      SELECT shareid FROM loghubods.user_share_log_per5min
+      WHERE topic='click'
+    )
+    SELECT * FROM source JOIN click USING (shareid)
+    """
+    with pytest.raises(SQLValidationError, match="user_share_log_per5min"):
+        SQLGuard.validate_partition_predicates(
+            sql, {"loghubods.user_share_log_per5min": ["dt"]}
+        )
+
+
 def test_video_action_applet_partition_requires_dt_but_not_business() -> None:
     partitions = {"loghubods.video_action_log_applet": ["dt", "business"]}
     SQLGuard.validate_partition_predicates(
@@ -90,10 +131,37 @@ def test_offline_product_efficiency_requires_extparams_root_session() -> None:
     SQLGuard.validate_product_efficiency_contract(good, "offline")
 
 
+def test_offline_total_only_product_efficiency_does_not_require_bucketing() -> None:
+    sql = """
+    SELECT COUNT(DISTINCT machinecode) AS dau
+    FROM loghubods.useractive_log
+    WHERE dt='20260810' AND apptype='0' AND businesstype='path'
+    """
+    SQLGuard.validate_product_efficiency_contract(sql, "offline", bucketed=False)
+
+
 def test_realtime_product_efficiency_does_not_use_offline_contract() -> None:
     SQLGuard.validate_product_efficiency_contract("SELECT 1", "realtime")
 
 
+def test_realtime_product_efficiency_per5min_sources_require_day_prefix() -> None:
+    invalid = """
+    WITH dau AS (
+      SELECT COUNT(*) FROM loghubods.useractive_log_per5min
+      WHERE dt='20260812'
+    ), shares AS (
+      SELECT COUNT(*) FROM loghubods.user_share_log_per5min
+      WHERE dt='20260812'
+    )
+    SELECT * FROM dau CROSS JOIN shares
+    """
+    with pytest.raises(SQLValidationError, match="dt LIKE"):
+        SQLGuard.validate_product_efficiency_contract(invalid, "realtime")
+
+    valid = invalid.replace("dt='20260812'", "dt LIKE '20260812%'")
+    SQLGuard.validate_product_efficiency_contract(valid, "realtime")
+
+
 def test_realtime_product_efficiency_rejects_video_business_filter() -> None:
     sql = """
     SELECT mid, GET_JSON_OBJECT(extparams, '$.rootSessionId') root_session_id