feat: 补齐 #1390 P3 决策信号生命周期与去重 (#1692)

* Implement decision signal lifecycle deduplication

* Fix API router prefix compatibility

* Fix null decision signal expiry defaults

api/app.py

@@ -265,7 +265,7 @@ def create_app(static_dir: Optional[Path] = None) -> FastAPI:
     # 注册路由
     # ============================================================
     
-    app.include_router(api_v1_router)
+    app.include_router(api_v1_router, prefix="/api/v1")
     add_error_handlers(app)
     
     # ============================================================

api/v1/endpoints/decision_signals.py

@@ -75,16 +75,18 @@ def _internal_error(message: str, exc: Exception) -> HTTPException:
     },
     summary="创建或去重决策信号",
     description=(
-        "显式写入 DecisionSignal。命中同源去重键时返回已有记录和 created=false；"
-        "若已有记录为 expired 且新请求为 active 并携带未来 expires_at，则原地刷新该记录；"
-        "P1 不保证并发绝对幂等。"
+        "显式写入 DecisionSignal。未传 horizon/expires_at 时由服务补默认生命周期；"
+        "命中同源去重键或窄 relaxed 去重时返回已有记录和 created=false；"
+        "active 新建或 expired 续期会失效同股旧 active 相反信号，"
+        "active duplicate retry 也会重跑该修复；普通旧 duplicate/replay 不作为新的激活事件；"
+        "不保证并发绝对幂等。"
     ),
     operation_id="createDecisionSignal",
 )
 def create_signal(request: DecisionSignalCreateRequest) -> DecisionSignalMutationResponse:
     service = DecisionSignalService()
     try:
-        payload = request.model_dump()
+        payload = request.model_dump(exclude_unset=True)
         return DecisionSignalMutationResponse(**service.create_signal(payload))
     except DecisionSignalStorageError as exc:
         raise _internal_error("Create decision signal failed", exc)
@@ -234,7 +236,10 @@ def get_signal(signal_id: int) -> DecisionSignalItem:
         500: {"model": ErrorResponse, "description": "更新失败"},
     },
     summary="更新决策信号状态",
-    description="只更新合法状态和可选 metadata；传入 metadata 时按整包替换保存，P1 不实现复杂状态机。",
+    description=(
+        "只更新合法状态和可选 metadata；传入 metadata 时按整包替换保存。"
+        "expired/invalidated/closed/archived 等 terminal 状态不能直接 PATCH 回 active。"
+    ),
     operation_id="updateDecisionSignalStatus",
 )
 def update_status(signal_id: int, request: DecisionSignalStatusUpdateRequest) -> DecisionSignalItem:

api/v1/router.py

@@ -27,8 +27,9 @@ from api.v1.endpoints import (
     usage,
 )
 
-# 创建 v1 版本主路由
-router = APIRouter(prefix="/api/v1")
+# 创建 v1 版本主路由。
+# /api/v1 前缀在 api.app 挂载，避免新版 FastAPI 误判子路由 "" 为 empty path。
+router = APIRouter()
 
 router.include_router(
     auth.router,

docs/CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [改进] #1390 P0 为个股分析与历史/回测展示新增可选八态 `action` / `action_label` 建议动作字段，保留 `operation_advice` 自由文本和 `decision_type=buy|hold|sell` 统计口径，不新增迁移或配置项。
 - [新功能] #1390 P1 新增独立 `DecisionSignal` 存储、Repository、Service 与 `/api/v1/decision-signals` API，支持按来源类型/市场/股票/动作/期限/阶段去重、按 `source_report_id` / `trace_id` 查询、同源过期信号续期且保留来源身份字段、禁止 expired 直接 PATCH 复活、价格计划校验、状态更新、懒过期、cache-only 持仓过滤、敏感信息脱敏、敏感 `trace_id` 拒绝和仅清理 `source_type=analysis` 历史绑定信号的历史删除联动。
 - [改进] #1390 P1 补充 Web decision-signals typed API wrapper 与契约隔离测试，暂不接入 UI。
+- [改进] #1390 P3 为 `DecisionSignal` 补齐默认生命周期、同源窄 relaxed 去重、相反 active 信号自动 invalidated、terminal 状态不可 PATCH 复活和自动提取低敏 market phase hints，保持 API 响应 schema 不变。
 - [修复] #1390 收紧建议动作 legacy fallback：英文 `not to ...` 与 `avoid selling/reducing/trimming ...` 等否定/回避表达不再误判为买卖动作，Web 旧记录不再把中文金融上下文、`buy or sell`、多 guard 歧义文本或 `buyback` / `buy-back` / `buy back` / `selloff` / `sell-off` / `sell off` 等英文复合词渲染成 action badge，并在有结构化 `action` 时让回测/历史趋势等入口按界面语言显示 action 标签。
 - [改进] 完善运行时日志上下文，补充 logger name、触发来源、市场统计与实时行情预取链路状态，便于排查调度、API、Bot 和数据源降级路径。
 - [新功能] 新增分析任务与历史报告运行流快照 API，提供 lanes、nodes、edges、events、summary 等统一契约，并从任务队列、运行诊断和 AnalysisContextPack overview 构建脱敏数据流/信息流。

docs/architecture/api_spec.json

@@ -778,7 +778,7 @@
           "DecisionSignals"
         ],
         "summary": "创建或去重决策信号",
-        "description": "显式写入 DecisionSignal。命中同源去重键时返回已有记录和 created=false；若已有记录为 expired 且新请求为 active 并携带未来 expires_at，则原地刷新该记录；P1 不保证并发绝对幂等。",
+        "description": "显式写入 DecisionSignal。未传 horizon/expires_at 时由服务补默认生命周期；命中同源去重键或窄 relaxed 去重时返回已有记录和 created=false；active 新建或 expired 续期会失效同股旧 active 相反信号，active duplicate retry 也会重跑该修复；普通旧 duplicate/replay 不作为新的激活事件；不保证并发绝对幂等。",
         "operationId": "createDecisionSignal",
         "security": [
           {
@@ -1395,7 +1395,7 @@
           "DecisionSignals"
         ],
         "summary": "更新决策信号状态",
-        "description": "只更新合法状态和可选 metadata；传入 metadata 时按整包替换保存，P1 不实现复杂状态机。",
+        "description": "只更新合法状态和可选 metadata；传入 metadata 时按整包替换保存。expired/invalidated/closed/archived 等 terminal 状态不能直接 PATCH 回 active。",
         "operationId": "updateDecisionSignalStatus",
         "security": [
           {

docs/full-guide.md

@@ -1302,7 +1302,7 @@ python main.py --debug
 
 #1390 P0 不会把后续信号资产字段平铺到现有 summary、历史列表、StockBar 或回测响应。#1390 P1 开始通过独立 `DecisionSignal` 资源承接 `horizon`、`plan_quality`、`status` 等更细粒度计划字段，仍不改变既有报告主契约、不回填历史、不新增配置项。
 
-### 决策信号资产（#1390 P1/P2）
+### 决策信号资产（#1390 P1/P2/P3）
 
 `DecisionSignal` 是独立后端资源，用于把 AI 建议沉淀为可查询、可去重、可更新状态的信号资产。它不替换 `operation_advice`、不扩展 `decision_type=buy|hold|sell`。#1390 P2 开始，普通个股分析和 Agent 个股分析在分析历史保存成功后，会从最终 `AnalysisResult` best-effort 提取一条 `source_type=analysis` 的信号；显式 API 或 service 调用仍然保留。
 
@@ -1310,19 +1310,21 @@ python main.py --debug
 
 P2 自动提取的市场阶段优先读取保存快照中的 `market_phase_summary.phase`，其次读取 `AnalysisResult.market_phase_summary.phase`；数据质量优先读取保存快照中的 `analysis_context_pack_overview.data_quality`，其次读取 `AnalysisResult.analysis_context_pack_overview.data_quality`。价格计划复用历史保存的狙击点解析规则，从 `dashboard.battle_plan.sniper_points.ideal_buy/secondary_buy/stop_loss/take_profit` 映射到 `entry_low/entry_high/stop_loss/target_price`；只有 `ideal_buy` 时写入 `entry_low`，只有 `secondary_buy` 时写入 `entry_high`，两者同时存在时按有效价格排序为 `entry_low <= entry_high`。缺失止损或目标价只会降低 service 自动计算的 `plan_quality`，不会编造字段。`watch_conditions` 优先读取 `dashboard.phase_decision.watch_conditions`，没有时才读取 `dashboard.battle_plan.action_checklist`；`catalyst_summary` 仅在 `dashboard.intelligence.positive_catalysts` 存在且为列表时写入。`confidence` 由报告置信等级做保守映射：`高/high=0.8`、`中/medium/mid=0.6`、`低/low=0.4`，原始置信等级保留在 `metadata`。
 
+P3 开始，生命周期由 `DecisionSignalService` 统一补齐：显式传入的 `horizon` / `expires_at` 永远优先；未传 `horizon` 时，`alert` 或 `premarket/intraday/lunch_break/closing_auction` 默认 `intraday`，`postmarket/non_trading/unknown` 或无阶段上下文时默认 `3d`；未传 `expires_at` 时，`intraday` 优先读取 `metadata.market_phase_summary.minutes_to_close/minutes_to_open`，无上下文时使用确定性 TTL fallback（A 股 4h、港股 5.5h、美股 6.5h、未知 4h），`1d/3d/5d/10d` 按自然日，`swing/long` 不自动过期。fallback TTL 只是缺少交易日历上下文时的降级策略，不等价于真实交易所收盘时间。自动提取只把 `market_phase_summary.phase/session_date/minutes_to_open/minutes_to_close` 作为低敏 hint 写入 `metadata.market_phase_summary`，最终 `horizon/expires_at` 仍由 service 计算。
+
 核心字段包括 `stock_code`、`stock_name`、`market`、`source_type`、`source_agent`、`source_report_id`、`trace_id`、`market_phase`、`trigger_source`、`action`、`action_label`、`confidence`、`score`、`horizon`、`entry_low`、`entry_high`、`stop_loss`、`target_price`、`invalidation`、`watch_conditions`、`reason`、`risk_summary`、`catalyst_summary`、`evidence`、`data_quality_summary`、`plan_quality`、`status`、`expires_at`、`created_at`、`updated_at` 和 `metadata`。`action` 复用八态建议动作；`market_phase` 复用市场阶段枚举；`source_type` 支持 `analysis|agent|alert|market_review|manual`；`status` 支持 `active|expired|invalidated|closed|archived`；`horizon` 支持 `intraday|1d|3d|5d|10d|swing|long`。
 
 `confidence` 为 `0.0-1.0`，`score` 为 `0-100`，与历史报告的 `sentiment_score` 解耦。价格计划字段 `entry_low`、`entry_high`、`stop_loss`、`target_price` 必须是有限正数，且同时传入 `entry_low` 和 `entry_high` 时要求 `entry_low <= entry_high`。`plan_quality` 支持 `complete|partial|minimal|unknown`：调用方显式传入合法值时直接保存；未传时由 service 计算，入场区间（`entry_low` 或 `entry_high` 任一有值）算 1 项，`stop_loss`、`target_price`、`invalidation`、`watch_conditions` 各算 1 项，满足 2 项为 `partial`，满足 4 项及以上为 `complete`，仅有 action/reason 为 `minimal`。
 
 新增 API：
 
-- `POST /api/v1/decision-signals`：创建或按同源键去重，返回 `{ item, created }`，HTTP 200。去重键为 `(source_report_id, source_type, market, stock_code, action, horizon, market_phase)`；没有 report 但有 `trace_id` 时使用 `(trace_id, source_type, market, stock_code, action, horizon, market_phase)`；两者皆无则不去重。`source_type` 是来源命名空间，manual/pre-report 弱引用不会与真实 analysis 绑定信号互相去重；`horizon` 和 `market_phase` 同为 `NULL` 时才互相去重，不同来源类型、不同市场、不同期限或不同市场阶段允许保存多条信号。若命中同源 expired 记录，且新请求为 active 并携带未来 `expires_at`，会原地刷新该记录并返回 `created=false`。P1 不提供并发唯一性保证。
+- `POST /api/v1/decision-signals`：创建或按同源键去重，返回 `{ item, created }`，HTTP 200。精确去重键为 `(source_report_id, source_type, market, stock_code, action, horizon, market_phase)`；没有 report 但有 `trace_id` 时使用 `(trace_id, source_type, market, stock_code, action, horizon, market_phase)`；两者皆无则不去重。精确匹配失败后，会按同源 + `source_type/market/stock_code/action` 做窄 relaxed fallback，只填补旧记录为空的 `horizon/market_phase`，且 `horizon` 只有在新值由 service 默认生成时才可填补；显式不同期限或已有不同阶段仍保留多条。若命中同源 expired 记录，且新请求为 active 并携带未来 `expires_at`，会原地刷新该记录并返回 `created=false`，这次续期按新的 active 激活事件处理。active 新建或 expired 续期后的 bullish 信号（`buy/add`）会把更早的 active defensive 信号（`reduce/sell/avoid`）标记为 `invalidated`，反向同理；active duplicate retry 也会重跑该失效修复，以恢复上次创建成功但失效写入失败的 partial create；普通旧 duplicate/replay 不作为新的激活事件。`hold/watch/alert` 不触发自动失效。API 响应 schema 不变，刷新或重复命中都对外返回 `created=false`；本功能不提供并发唯一性保证。
 - `GET /api/v1/decision-signals`：分页查询，支持 `market`、`stock_code`、`action`、`market_phase`、`source_type`、`source_report_id`、`trace_id`、`trigger_source`、`status`、时间范围、`holding_only`、`account_id`。
 - `GET /api/v1/decision-signals/{signal_id}`：查询单条，不存在返回 404。
-- `PATCH /api/v1/decision-signals/{signal_id}/status`：更新合法状态和可选 `metadata`；传入 `metadata` 时按整包替换保存，不实现复杂状态机。
+- `PATCH /api/v1/decision-signals/{signal_id}/status`：更新合法状态和可选 `metadata`；传入 `metadata` 时按整包替换保存。`expired/invalidated/closed/archived` 等 terminal 状态不能直接 PATCH 回 `active`，expired 续期仍只能重新 `POST` active + 未来 `expires_at`。
 - `GET /api/v1/decision-signals/latest/{stock_code}`：按股票查询最新 active 信号，默认 `limit=1`。
 
-读取入口会懒过期：列表、详情和 latest 查询前会把已到 `expires_at` 的 active 信号标为 expired；创建时已过期的 active 信号会直接保存为 expired；同源 expired 信号只能通过重新 `POST` active + 未来 `expires_at` 的方式延展，`PATCH /status` 不接受 `expires_at`。`closed|invalidated|archived` 不会被 create 路径复活。时间字段按 UTC 归一化为无时区 `datetime` 保存和比较；带时区输入会先转为 UTC 后去掉 `tzinfo`，无时区输入按 UTC 处理，API 响应继续返回不带时区后缀的 ISO 字符串。股票代码入库与查询按 `market` 确定性归一化：A 股 `600519`、`SH600519`、`600519.SH` 等常见变体按同一代码匹配；港股 `00700`、`HK00700`、`00700.HK` 按 `HK00700` 匹配；美股 ticker 统一大写。`holding_only=true` 只读取 active 账户下 `portfolio_positions` 中 `quantity > 0` 的缓存持仓，并按持仓 `(market, stock_code)` 匹配信号，可选 active `account_id`；该查询不会调用组合 snapshot replay，无缓存时返回空结果，需先通过 portfolio snapshot API 刷新缓存。
+读取入口会懒过期：列表、详情和 latest 查询前会把已到 `expires_at` 的 active 信号标为 expired；创建时已过期的 active 信号会直接保存为 expired；同源 expired 信号只能通过重新 `POST` active + 未来 `expires_at` 的方式延展，`PATCH /status` 不接受 `expires_at`。`expired|invalidated|closed|archived` 不会被 PATCH 直接复活，`closed|invalidated|archived` 也不会被 create 路径复活。相反信号自动失效会合并写入旧信号 `metadata`：`invalidated_by_signal_id`、`invalidated_reason`、`invalidated_at`、`previous_status`；旧 metadata JSON 损坏时会替换为失效 metadata 并写入 `metadata_replaced_due_to_invalid_json=true`，不阻断新信号创建。时间字段按 UTC 归一化为无时区 `datetime` 保存和比较；带时区输入会先转为 UTC 后去掉 `tzinfo`，无时区输入按 UTC 处理，API 响应继续返回不带时区后缀的 ISO 字符串。股票代码入库与查询按 `market` 确定性归一化：A 股 `600519`、`SH600519`、`600519.SH` 等常见变体按同一代码匹配；港股 `00700`、`HK00700`、`00700.HK` 按 `HK00700` 匹配；美股 ticker 统一大写。`holding_only=true` 只读取 active 账户下 `portfolio_positions` 中 `quantity > 0` 的缓存持仓，并按持仓 `(market, stock_code)` 匹配信号，可选 active `account_id`；该查询不会调用组合 snapshot replay，无缓存时返回空结果，需先通过 portfolio snapshot API 刷新缓存。
 
 `source_report_id` 可为空且不强制校验历史记录存在；删除历史记录时只显式清理 `source_type=analysis` 且 `source_report_id` 命中实际删除 ID 的历史绑定信号，`manual/agent/alert/market_review` 等弱引用信号不会仅因 ID 碰撞被删除；列表接口支持按 `source_report_id` 和 `trace_id` 做 typed filter。`task_id`、`alert_trigger_id` 等后续关联字段先放入 `metadata`，P1 不新增独立列，也不提供 typed filter，后续联动阶段再提升为独立契约。JSON 字段、长文本字段和展示型短文本字段（`stock_name/source_agent/trigger_source/action_label`）会在写入前执行信号专用脱敏，覆盖敏感 key、Bearer、Authorization/Cookie header 或赋值、token-like 字符串、其他敏感赋值、webhook URL、URL userinfo 以及带敏感 query/fragment 参数的 URL；普通证据 URL 会保留以保证来源可追溯，且长文本不会套用诊断文本的 300 字符截断。`trace_id` 是同源去重身份字段，若包含会被脱敏的敏感 credential，API 会拒绝请求而不是保存有损 redaction 后的值。
 

docs/full-guide_EN.md

@@ -1131,27 +1131,29 @@ Unknown or ambiguous advice is not coerced into `watch` or `hold`; it returns em
 
 #1390 P0 does not flatten future signal-asset fields into current report summaries, history lists, StockBar rows, or backtest responses. #1390 P1 now carries more granular plan fields such as `horizon`, `plan_quality`, and `status` through an independent `DecisionSignal` resource; it still does not change the existing report contract, backfill history, or add configuration.
 
-### Decision Signal Asset (#1390 P1/P2)
+### Decision Signal Asset (#1390 P1/P2/P3)
 
-`DecisionSignal` is an independent backend resource for persisting AI recommendations as queryable, deduplicated, status-updatable signal assets. It does not replace `operation_advice` or expand the legacy `decision_type=buy|hold|sell` contract. Starting with #1390 P2, regular stock analysis and Agent stock analysis best-effort extract one `source_type=analysis` signal from the final `AnalysisResult` after analysis history is saved successfully; explicit API and service calls remain supported.
+`DecisionSignal` is an independent backend resource for persisting AI recommendations as queryable, deduplicated, status-updatable signal assets. It does not replace `operation_advice` or expand the legacy `decision_type=buy|hold|sell` contract. Starting with #1390 P2, regular stock analysis and Agent stock analysis best-effort extract one `source_type=analysis` signal from the final `AnalysisResult` after analysis history is saved successfully; explicit API and service calls remain supported. #1390 P3 adds default lifecycle handling, narrow same-source relaxed deduplication, opposite-signal invalidation, and stricter terminal-state transitions without changing the public response schema.
 
 Automatic extraction consumes structured fields from the completed report only. It does not parse Markdown, backfill old history, add configuration, or change the main report contract. Extraction failures, unknown or ambiguous advice, non-stock reports, and unrecognized markets skip signal writes without affecting report persistence. `source_report_id` is the just-saved `AnalysisHistory.id`; `trace_id` prefers the runtime diagnostics trace and falls back to the pipeline trace or `query_id`; `stock_name` comes from `AnalysisResult.name`; `trigger_source` comes from the runtime entrypoint and falls back to `system`.
 
 For P2 automatic extraction, `market_phase` first reads `market_phase_summary.phase` from the saved context snapshot and then falls back to `AnalysisResult.market_phase_summary.phase`; data quality first reads `analysis_context_pack_overview.data_quality` from the saved context snapshot and then falls back to `AnalysisResult.analysis_context_pack_overview.data_quality`. Price-plan extraction reuses the same sniper-point parser used by history persistence, mapping `dashboard.battle_plan.sniper_points.ideal_buy/secondary_buy/stop_loss/take_profit` to `entry_low/entry_high/stop_loss/target_price`; `ideal_buy` alone writes `entry_low`, `secondary_buy` alone writes `entry_high`, and when both are present they are sorted into `entry_low <= entry_high`. Missing stop-loss or target prices only lower the service-computed `plan_quality` instead of inventing fields. `watch_conditions` first reads `dashboard.phase_decision.watch_conditions` and then falls back to `dashboard.battle_plan.action_checklist`. `catalyst_summary` is written only when `dashboard.intelligence.positive_catalysts` exists and is a list. `confidence` uses a conservative report-level mapping: `高/high=0.8`, `中/medium/mid=0.6`, `低/low=0.4`; the original report confidence level remains in `metadata`.
 
+Starting with P3, `DecisionSignalService` owns lifecycle defaults. Explicit `horizon` / `expires_at` values always win. When `horizon` is omitted, `alert` or `premarket/intraday/lunch_break/closing_auction` defaults to `intraday`, while `postmarket/non_trading/unknown` or missing phase context defaults to `3d`. When `expires_at` is omitted, `intraday` first uses `metadata.market_phase_summary.minutes_to_close/minutes_to_open`; without context it uses deterministic TTL fallback values (CN 4h, HK 5.5h, US 6.5h, unknown 4h). `1d/3d/5d/10d` use natural days, and `swing/long` do not auto-expire. The fallback TTL is only a no-context degradation path, not an exchange-calendar close time. Automatic extraction writes only low-sensitive `market_phase_summary.phase/session_date/minutes_to_open/minutes_to_close` hints into `metadata.market_phase_summary`; final `horizon/expires_at` values are still computed by the service.
+
 Core fields include `stock_code`, `stock_name`, `market`, `source_type`, `source_agent`, `source_report_id`, `trace_id`, `market_phase`, `trigger_source`, `action`, `action_label`, `confidence`, `score`, `horizon`, `entry_low`, `entry_high`, `stop_loss`, `target_price`, `invalidation`, `watch_conditions`, `reason`, `risk_summary`, `catalyst_summary`, `evidence`, `data_quality_summary`, `plan_quality`, `status`, `expires_at`, `created_at`, `updated_at`, and `metadata`. `action` reuses the eight-state action taxonomy; `market_phase` reuses the market phase enum; `source_type` supports `analysis|agent|alert|market_review|manual`; `status` supports `active|expired|invalidated|closed|archived`; `horizon` supports `intraday|1d|3d|5d|10d|swing|long`.
 
 `confidence` is `0.0-1.0`, and `score` is `0-100`, separate from historical `sentiment_score`. Price-plan fields `entry_low`, `entry_high`, `stop_loss`, and `target_price` must be finite positive numbers; when both `entry_low` and `entry_high` are present, `entry_low <= entry_high` is required. `plan_quality` supports `complete|partial|minimal|unknown`: a valid explicit value is saved as-is; otherwise the service computes it. The entry range (`entry_low` or `entry_high`) counts as one slot, and `stop_loss`, `target_price`, `invalidation`, and `watch_conditions` each count as one slot. Two slots produce `partial`, four or more produce `complete`, and action/reason without enough slots produces `minimal`.
 
 New API endpoints:
 
-- `POST /api/v1/decision-signals`: create or deduplicate a signal and return `{ item, created }` with HTTP 200. Deduplication uses `(source_report_id, source_type, market, stock_code, action, horizon, market_phase)` when `source_report_id` is present, or `(trace_id, source_type, market, stock_code, action, horizon, market_phase)` when only `trace_id` is present. Signals without either source identifier are not deduplicated. `source_type` is a source namespace, so manual/pre-report weak references do not deduplicate against real analysis-bound signals. `NULL` `horizon` and `NULL` `market_phase` deduplicate only against the same `NULL` dimensions; different source types, markets, horizons, or market phases may persist as separate signals. When the same source key matches an expired signal and the new request is active with a future `expires_at`, the existing row is refreshed in place and still returns `created=false`. P1 does not guarantee concurrent idempotency.
+- `POST /api/v1/decision-signals`: create or deduplicate a signal and return `{ item, created }` with HTTP 200. Exact deduplication uses `(source_report_id, source_type, market, stock_code, action, horizon, market_phase)` when `source_report_id` is present, or `(trace_id, source_type, market, stock_code, action, horizon, market_phase)` when only `trace_id` is present. Signals without either source identifier are not deduplicated. After an exact miss, a narrow relaxed fallback searches the same source plus `source_type/market/stock_code/action` and only fills old blank `horizon/market_phase` values. `horizon` can be filled only when the new value was generated by the service default; explicit different horizons or already different phases remain separate rows. When the same source key matches an expired signal and the new request is active with a future `expires_at`, the existing row is refreshed in place, still returns `created=false`, and that renewal is treated as a new active activation event. Active creation or expired renewal of a bullish signal (`buy/add`) invalidates earlier active defensive signals (`reduce/sell/avoid`) for the same stock, and the reverse also applies; active duplicate retries also rerun this repair to recover from a previous partial create where the signal was saved but invalidation failed; ordinary old duplicate/replay attempts are not treated as new activation events. `hold/watch/alert` do not trigger automatic invalidation. The API response schema is unchanged, and both refreshed and duplicate outcomes return `created=false`. P3 does not guarantee concurrent idempotency.
 - `GET /api/v1/decision-signals`: paginated query with `market`, `stock_code`, `action`, `market_phase`, `source_type`, `source_report_id`, `trace_id`, `trigger_source`, `status`, time ranges, `holding_only`, and `account_id`.
 - `GET /api/v1/decision-signals/{signal_id}`: fetch one signal; missing IDs return 404.
-- `PATCH /api/v1/decision-signals/{signal_id}/status`: update a valid status and optional `metadata`; when `metadata` is provided it replaces the whole stored metadata object, and no complex state machine is enforced.
+- `PATCH /api/v1/decision-signals/{signal_id}/status`: update a valid status and optional `metadata`; when `metadata` is provided it replaces the whole stored metadata object. `expired/invalidated/closed/archived` terminal states cannot be patched directly back to `active`; expired renewal still requires re-posting active data with a future `expires_at`.
 - `GET /api/v1/decision-signals/latest/{stock_code}`: return latest active signals for a stock, default `limit=1`.
 
-Read paths lazily expire active signals whose `expires_at` has passed before list, detail, and latest queries; creating an already expired active signal stores it as `expired`; the same-source expired signal can only be extended by re-posting active data with a future `expires_at`, and `PATCH /status` does not accept `expires_at`. `closed|invalidated|archived` signals are not reactivated by the create path. Time fields are normalized to UTC naive datetimes for storage and comparison; timezone-aware inputs are converted to UTC and stripped of `tzinfo`, naive inputs are treated as UTC, and API responses continue to return ISO strings without timezone suffixes. Stock codes are normalized deterministically by `market`: CN variants such as `600519`, `SH600519`, and `600519.SH` match the same stored code; HK variants such as `00700`, `HK00700`, and `00700.HK` match `HK00700`; US tickers are uppercased. `holding_only=true` reads only cached `portfolio_positions` rows with `quantity > 0` under active accounts and matches signals by the held `(market, stock_code)`, optionally scoped by an active `account_id`; it does not call portfolio snapshot replay. When no cache exists, it returns an empty result and callers should refresh the cache through the portfolio snapshot API first.
+Read paths lazily expire active signals whose `expires_at` has passed before list, detail, and latest queries; creating an already expired active signal stores it as `expired`; the same-source expired signal can only be extended by re-posting active data with a future `expires_at`, and `PATCH /status` does not accept `expires_at`. `expired|invalidated|closed|archived` cannot be patched directly back to active, and `closed|invalidated|archived` are not reactivated by the create path. Automatic opposite-signal invalidation merges these fields into the old signal metadata: `invalidated_by_signal_id`, `invalidated_reason`, `invalidated_at`, and `previous_status`. If old metadata JSON is corrupt, it is replaced with invalidation metadata plus `metadata_replaced_due_to_invalid_json=true`, and the new signal creation is not blocked. Time fields are normalized to UTC naive datetimes for storage and comparison; timezone-aware inputs are converted to UTC and stripped of `tzinfo`, naive inputs are treated as UTC, and API responses continue to return ISO strings without timezone suffixes. Stock codes are normalized deterministically by `market`: CN variants such as `600519`, `SH600519`, and `600519.SH` match the same stored code; HK variants such as `00700`, `HK00700`, and `00700.HK` match `HK00700`; US tickers are uppercased. `holding_only=true` reads only cached `portfolio_positions` rows with `quantity > 0` under active accounts and matches signals by the held `(market, stock_code)`, optionally scoped by an active `account_id`; it does not call portfolio snapshot replay. When no cache exists, it returns an empty result and callers should refresh the cache through the portfolio snapshot API first.
 
 `source_report_id` is nullable and is not required to reference an existing history row; deleting history records explicitly removes only history-bound signals with `source_type=analysis` whose `source_report_id` matches actually deleted IDs, so `manual/agent/alert/market_review` weak-reference signals are not deleted solely because of an ID collision. The list endpoint supports typed filters for `source_report_id` and `trace_id`. Follow-up association fields such as `task_id` and `alert_trigger_id` should be stored in `metadata` for P1; P1 does not add dedicated columns or typed filters for them, which are deferred to the later integration phase. JSON fields, long text fields, and public short text fields (`stock_name/source_agent/trigger_source/action_label`) are sanitized before persistence with a signal-specific sanitizer that redacts sensitive keys, Bearer values, Authorization/Cookie headers or assignments, token-like strings, other sensitive assignments, webhook URLs, URL userinfo, and URLs with sensitive query or fragment parameters. Ordinary evidence URLs are preserved for source traceability, and long text does not use the diagnostics 300-character truncation. `trace_id` is a same-source identity field; if it contains sensitive credentials that would be redacted, the API rejects the request instead of storing a lossy redacted value.
 

src/repositories/decision_signal_repo.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -16,9 +17,25 @@ from src.storage import (
 )
 
 
+@dataclass
+class DecisionSignalCreateResult:
+    """Outcome of an idempotent DecisionSignal create attempt."""
+
+    row: DecisionSignalRecord
+    created: bool
+    refreshed: bool = False
+    duplicate: bool = False
+    invalidation_reference_at: Optional[datetime] = None
+
+    def __iter__(self):
+        yield self.row
+        yield self.created
+
+
 class DecisionSignalRepository:
     """DB access layer for persisted AI decision signals."""
 
+    _RELAXED_MERGE_STATUSES = frozenset({"active", "expired"})
     _IMMUTABLE_REFRESH_FIELDS = frozenset({
         "id",
         "created_at",
@@ -46,7 +63,12 @@ class DecisionSignalRepository:
             session.refresh(row)
             return row
 
-    def create_if_absent(self, fields: Dict[str, Any]) -> Tuple[DecisionSignalRecord, bool]:
+    def create_if_absent(
+        self,
+        fields: Dict[str, Any],
+        *,
+        allow_relaxed_horizon_fill: bool = False,
+    ) -> DecisionSignalCreateResult:
         self.expire_due_signals()
         fields = self._normalize_datetime_fields(fields)
         with self.db.get_session() as session:
@@ -56,12 +78,71 @@ class DecisionSignalRepository:
                     self._refresh_existing_in_session(existing, fields)
                     session.commit()
                     session.refresh(existing)
-                return existing, False
+                    return DecisionSignalCreateResult(
+                        row=existing,
+                        created=False,
+                        refreshed=True,
+                        invalidation_reference_at=existing.updated_at,
+                    )
+                return DecisionSignalCreateResult(
+                    row=existing,
+                    created=False,
+                    duplicate=True,
+                    invalidation_reference_at=existing.created_at,
+                )
+
+            relaxed_existing = self._find_relaxed_existing_in_session(
+                session=session,
+                fields=fields,
+                allow_relaxed_horizon_fill=allow_relaxed_horizon_fill,
+            )
+            if relaxed_existing is not None:
+                if self._should_refresh_existing(relaxed_existing, fields):
+                    self._refresh_existing_in_session(relaxed_existing, fields)
+                    self._fill_relaxed_dimensions_in_session(
+                        relaxed_existing,
+                        fields,
+                        allow_horizon_fill=allow_relaxed_horizon_fill,
+                    )
+                    session.commit()
+                    session.refresh(relaxed_existing)
+                    return DecisionSignalCreateResult(
+                        row=relaxed_existing,
+                        created=False,
+                        refreshed=True,
+                        invalidation_reference_at=relaxed_existing.updated_at,
+                    )
+                if relaxed_existing.status == "active":
+                    changed = self._fill_relaxed_dimensions_in_session(
+                        relaxed_existing,
+                        fields,
+                        allow_horizon_fill=allow_relaxed_horizon_fill,
+                    )
+                    if changed:
+                        session.commit()
+                        session.refresh(relaxed_existing)
+                        return DecisionSignalCreateResult(
+                            row=relaxed_existing,
+                            created=False,
+                            refreshed=True,
+                            invalidation_reference_at=relaxed_existing.created_at,
+                        )
+                return DecisionSignalCreateResult(
+                    row=relaxed_existing,
+                    created=False,
+                    duplicate=True,
+                    invalidation_reference_at=relaxed_existing.created_at,
+                )
+
             row = DecisionSignalRecord(**fields)
             session.add(row)
             session.commit()
             session.refresh(row)
-            return row, True
+            return DecisionSignalCreateResult(
+                row=row,
+                created=True,
+                invalidation_reference_at=row.created_at,
+            )
 
     def get(self, signal_id: int) -> Optional[DecisionSignalRecord]:
         self.expire_due_signals()
@@ -155,6 +236,33 @@ class DecisionSignalRepository:
             ).scalars().all()
             return list(rows)
 
+    def list_active_by_stock_actions(
+        self,
+        *,
+        market: str,
+        stock_code: str,
+        actions: List[str],
+        exclude_signal_id: Optional[int] = None,
+    ) -> List[DecisionSignalRecord]:
+        self.expire_due_signals()
+        if not actions:
+            return []
+        conditions = [
+            DecisionSignalRecord.status == "active",
+            DecisionSignalRecord.market == market,
+            DecisionSignalRecord.stock_code == stock_code,
+            DecisionSignalRecord.action.in_(actions),
+        ]
+        if exclude_signal_id is not None:
+            conditions.append(DecisionSignalRecord.id != exclude_signal_id)
+        with self.db.get_session() as session:
+            rows = session.execute(
+                select(DecisionSignalRecord)
+                .where(and_(*conditions))
+                .order_by(desc(DecisionSignalRecord.created_at), desc(DecisionSignalRecord.id))
+            ).scalars().all()
+            return list(rows)
+
     def update_status(
         self,
         signal_id: int,
@@ -265,6 +373,90 @@ class DecisionSignalRepository:
             .limit(1)
         ).scalar_one_or_none()
 
+    @classmethod
+    def _find_relaxed_existing_in_session(
+        cls,
+        *,
+        session: Any,
+        fields: Dict[str, Any],
+        allow_relaxed_horizon_fill: bool,
+    ) -> Optional[DecisionSignalRecord]:
+        source_report_id = fields.get("source_report_id")
+        trace_id = fields.get("trace_id")
+        if source_report_id is None and not trace_id:
+            return None
+
+        conditions = [
+            DecisionSignalRecord.source_type == fields.get("source_type"),
+            DecisionSignalRecord.market == fields.get("market"),
+            DecisionSignalRecord.stock_code == fields.get("stock_code"),
+            DecisionSignalRecord.action == fields.get("action"),
+        ]
+        if source_report_id is not None:
+            conditions.append(DecisionSignalRecord.source_report_id == source_report_id)
+        else:
+            conditions.append(DecisionSignalRecord.trace_id == trace_id)
+
+        candidates = session.execute(
+            select(DecisionSignalRecord)
+            .where(and_(*conditions))
+            .order_by(DecisionSignalRecord.id.asc())
+        ).scalars().all()
+        for candidate in candidates:
+            if candidate.status not in cls._RELAXED_MERGE_STATUSES:
+                continue
+            if cls._can_relaxed_merge(
+                candidate,
+                fields,
+                allow_horizon_fill=allow_relaxed_horizon_fill,
+            ):
+                return candidate
+        return None
+
+    @classmethod
+    def _can_relaxed_merge(
+        cls,
+        existing: DecisionSignalRecord,
+        fields: Dict[str, Any],
+        *,
+        allow_horizon_fill: bool,
+    ) -> bool:
+        new_horizon = fields.get("horizon")
+        new_phase = fields.get("market_phase")
+
+        horizon_fill = existing.horizon is None and new_horizon is not None
+        if horizon_fill and not allow_horizon_fill:
+            return False
+        if existing.horizon is not None and existing.horizon != new_horizon:
+            return False
+
+        phase_fill = existing.market_phase is None and new_phase is not None
+        if existing.market_phase is not None and existing.market_phase != new_phase:
+            return False
+
+        return horizon_fill or phase_fill
+
+    @classmethod
+    def _fill_relaxed_dimensions_in_session(
+        cls,
+        existing: DecisionSignalRecord,
+        fields: Dict[str, Any],
+        *,
+        allow_horizon_fill: bool,
+    ) -> bool:
+        changed = False
+        new_horizon = fields.get("horizon")
+        new_phase = fields.get("market_phase")
+        if existing.horizon is None and new_horizon is not None and allow_horizon_fill:
+            existing.horizon = new_horizon
+            changed = True
+        if existing.market_phase is None and new_phase is not None:
+            existing.market_phase = new_phase
+            changed = True
+        if changed:
+            existing.updated_at = utc_naive_now()
+        return changed
+
     @staticmethod
     def _build_conditions(
         *,

src/services/decision_signal_extractor.py

@@ -66,6 +66,16 @@ def build_decision_signal_payload_from_report(
         sniper_points.get("secondary_buy"),
     )
 
+    metadata = {
+        "report_type": report_type,
+        "decision_type": getattr(result, "decision_type", None),
+        "report_confidence_level": getattr(result, "confidence_level", None),
+        "report_language": getattr(result, "report_language", None),
+    }
+    market_phase_summary = _extract_market_phase_summary(context_snapshot, result)
+    if market_phase_summary:
+        metadata["market_phase_summary"] = market_phase_summary
+
     payload: Dict[str, Any] = {
         "stock_code": raw_code,
         "stock_name": getattr(result, "name", None),
@@ -93,12 +103,7 @@ def build_decision_signal_payload_from_report(
         "watch_conditions": _watch_conditions(dashboard),
         "evidence": _evidence(result, sniper_points),
         "data_quality_summary": _extract_data_quality(context_snapshot, result),
-        "metadata": {
-            "report_type": report_type,
-            "decision_type": getattr(result, "decision_type", None),
-            "report_confidence_level": getattr(result, "confidence_level", None),
-            "report_language": getattr(result, "report_language", None),
-        },
+        "metadata": metadata,
         "report_language": getattr(result, "report_language", None),
     }
     return {key: value for key, value in payload.items() if value not in (None, "", [], {})}
@@ -183,6 +188,22 @@ def _extract_market_phase(context_snapshot: Optional[Mapping[str, Any]], result:
     return str(result_phase) if result_phase else None
 
 
+def _extract_market_phase_summary(
+    context_snapshot: Optional[Mapping[str, Any]],
+    result: AnalysisResult,
+) -> Optional[Dict[str, Any]]:
+    raw_summary = _as_mapping(_as_mapping(context_snapshot).get("market_phase_summary"))
+    if not raw_summary:
+        raw_summary = _as_mapping(getattr(result, "market_phase_summary", None))
+    allowed_fields = ("phase", "session_date", "minutes_to_open", "minutes_to_close")
+    summary = {
+        field_name: raw_summary.get(field_name)
+        for field_name in allowed_fields
+        if raw_summary.get(field_name) not in (None, "")
+    }
+    return summary or None
+
+
 def _extract_data_quality(context_snapshot: Optional[Mapping[str, Any]], result: AnalysisResult) -> Optional[Any]:
     snapshot_quality = _as_mapping(
         _as_mapping(context_snapshot).get("analysis_context_pack_overview")

src/services/decision_signal_service.py

@@ -6,7 +6,7 @@ from __future__ import annotations
 import json
 import logging
 import math
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional, Tuple, get_args
 
 from data_provider.base import canonical_stock_code, normalize_stock_code
@@ -32,6 +32,20 @@ HORIZONS = frozenset({"intraday", "1d", "3d", "5d", "10d", "swing", "long"})
 MARKET_PHASES = frozenset(phase.value for phase in MarketPhase)
 DECISION_ACTIONS = frozenset(get_args(DecisionAction))
 REDACTION_MARKERS = ("[REDACTED]", "[REDACTED_URL]")
+TERMINAL_STATUSES = frozenset({"expired", "invalidated", "closed", "archived"})
+BULLISH_ACTIONS = frozenset({"buy", "add"})
+DEFENSIVE_ACTIONS = frozenset({"reduce", "sell", "avoid"})
+INTRADAY_PHASES = frozenset({
+    MarketPhase.PREMARKET.value,
+    MarketPhase.INTRADAY.value,
+    MarketPhase.LUNCH_BREAK.value,
+    MarketPhase.CLOSING_AUCTION.value,
+})
+DEFAULT_INTRADAY_TTL_HOURS = {
+    "cn": 4.0,
+    "hk": 5.5,
+    "us": 6.5,
+}
 
 logger = logging.getLogger(__name__)
 
@@ -57,9 +71,18 @@ class DecisionSignalService:
         self.portfolio_repo = portfolio_repo or PortfolioRepository(db_manager)
 
     def create_signal(self, payload: Dict[str, Any]) -> Dict[str, Any]:
-        fields = self._normalize_payload(payload)
-        row, created = self.repo.create_if_absent(fields)
-        return {"item": self._serialize(row), "created": created}
+        fields, lifecycle = self._normalize_payload(payload)
+        result = self.repo.create_if_absent(
+            fields,
+            allow_relaxed_horizon_fill=lifecycle["horizon_defaulted"],
+        )
+        # Active duplicates can be retries after a prior partial create; rerun invalidation to repair old opposing signals.
+        if result.row.status == "active":
+            self._invalidate_opposing_active_signals(
+                result.row,
+                reference_at=result.invalidation_reference_at,
+            )
+        return {"item": self._serialize(result.row), "created": result.created}
 
     def get_signal(self, signal_id: int) -> Dict[str, Any]:
         row = self.repo.get(signal_id)
@@ -182,9 +205,9 @@ class DecisionSignalService:
         if existing is None:
             raise DecisionSignalNotFoundError(f"Decision signal not found: {signal_id}")
         if status_norm == "active" and (
-            existing.status == "expired" or self._is_expired(existing.expires_at)
+            existing.status in TERMINAL_STATUSES or self._is_expired(existing.expires_at)
         ):
-            raise ValueError("expired decision signal cannot be reactivated without extending expires_at")
+            raise ValueError("terminal decision signal cannot be reactivated through status update")
         row = self.repo.update_status(
             signal_id,
             status=status_norm,
@@ -195,7 +218,7 @@ class DecisionSignalService:
             raise DecisionSignalNotFoundError(f"Decision signal not found: {signal_id}")
         return self._serialize(row)
 
-    def _normalize_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+    def _normalize_payload(self, payload: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
         market = self._normalize_market(payload.get("market"))
         stock_code = self._normalize_stock_code(payload.get("stock_code"), market=market)
         action = self._normalize_action(payload.get("action"))
@@ -211,6 +234,22 @@ class DecisionSignalService:
         if score is not None and not 0 <= score <= 100:
             raise ValueError("score must be between 0 and 100")
 
+        market_phase = self._normalize_optional_enum(payload.get("market_phase"), MARKET_PHASES, "market_phase")
+        horizon_explicit = self._payload_has_value(payload, "horizon")
+        horizon = self._normalize_optional_enum(payload.get("horizon"), HORIZONS, "horizon")
+        horizon_defaulted = False
+        if horizon is None:
+            horizon = self._default_horizon(action=action, market_phase=market_phase)
+            horizon_defaulted = horizon is not None and not horizon_explicit
+        expires_explicit = self._payload_has_value(payload, "expires_at")
+        expires_at = self._parse_datetime(payload.get("expires_at"))
+        if expires_at is None and not expires_explicit:
+            expires_at = self._default_expires_at(
+                horizon=horizon,
+                market=market,
+                metadata=payload.get("metadata"),
+            )
+
         fields: Dict[str, Any] = {
             "stock_code": stock_code,
             "stock_name": self._optional_public_text(payload.get("stock_name"), "stock_name", max_length=64),
@@ -219,13 +258,13 @@ class DecisionSignalService:
             "source_agent": self._optional_public_text(payload.get("source_agent"), "source_agent", max_length=64),
             "source_report_id": self._optional_int(payload.get("source_report_id"), "source_report_id"),
             "trace_id": self._optional_identity_text(payload.get("trace_id"), "trace_id", max_length=64),
-            "market_phase": self._normalize_optional_enum(payload.get("market_phase"), MARKET_PHASES, "market_phase"),
+            "market_phase": market_phase,
             "trigger_source": self._normalize_trigger_source(payload.get("trigger_source")),
             "action": action,
             "action_label": action_label,
             "confidence": confidence,
             "score": score,
-            "horizon": self._normalize_optional_enum(payload.get("horizon"), HORIZONS, "horizon"),
+            "horizon": horizon,
             "entry_low": self._optional_price_float(payload.get("entry_low"), "entry_low"),
             "entry_high": self._optional_price_float(payload.get("entry_high"), "entry_high"),
             "stop_loss": self._optional_price_float(payload.get("stop_loss"), "stop_loss"),
@@ -238,7 +277,7 @@ class DecisionSignalService:
             "evidence_json": self._json_dumps(payload.get("evidence")),
             "data_quality_summary_json": self._json_dumps(payload.get("data_quality_summary")),
             "status": self._normalize_optional_enum(payload.get("status"), SIGNAL_STATUSES, "status") or "active",
-            "expires_at": self._parse_datetime(payload.get("expires_at")),
+            "expires_at": expires_at,
             "metadata_json": self._json_dumps(payload.get("metadata")),
         }
         if fields["status"] == "active" and self._is_expired(fields["expires_at"]):
@@ -248,7 +287,157 @@ class DecisionSignalService:
             payload.get("plan_quality"),
             fields=fields,
         )
-        return fields
+        return fields, {"horizon_defaulted": horizon_defaulted}
+
+    @staticmethod
+    def _payload_has_value(payload: Dict[str, Any], field_name: str) -> bool:
+        return payload.get(field_name) not in (None, "")
+
+    @staticmethod
+    def _default_horizon(*, action: str, market_phase: Optional[str]) -> str:
+        if action == "alert" or market_phase in INTRADAY_PHASES:
+            return "intraday"
+        return "3d"
+
+    @classmethod
+    def _default_expires_at(
+        cls,
+        *,
+        horizon: Optional[str],
+        market: str,
+        metadata: Any,
+    ) -> Optional[datetime]:
+        now = utc_naive_now()
+        if horizon == "intraday":
+            minutes_to_close = cls._metadata_minutes(metadata, "minutes_to_close")
+            if minutes_to_close is not None:
+                return now + timedelta(minutes=minutes_to_close)
+            minutes_to_open = cls._metadata_minutes(metadata, "minutes_to_open")
+            if minutes_to_open is not None:
+                fallback_minutes = int(cls._intraday_fallback_hours(market) * 60)
+                return now + timedelta(minutes=minutes_to_open + fallback_minutes)
+            return now + timedelta(hours=cls._intraday_fallback_hours(market))
+
+        days = cls._horizon_days(horizon)
+        if days is None:
+            return None
+        return now + timedelta(days=days)
+
+    @staticmethod
+    def _intraday_fallback_hours(market: str) -> float:
+        return DEFAULT_INTRADAY_TTL_HOURS.get(market, 4.0)
+
+    @staticmethod
+    def _horizon_days(horizon: Optional[str]) -> Optional[int]:
+        if horizon in {"1d", "3d", "5d", "10d"}:
+            return int(horizon[:-1])
+        return None
+
+    @classmethod
+    def _metadata_minutes(cls, metadata: Any, field_name: str) -> Optional[int]:
+        if not isinstance(metadata, dict):
+            return None
+        summary = metadata.get("market_phase_summary")
+        if not isinstance(summary, dict):
+            return None
+        value = summary.get(field_name)
+        if value in (None, ""):
+            return None
+        try:
+            minutes = int(float(value))
+        except (TypeError, ValueError):
+            return None
+        return minutes if minutes >= 0 else None
+
+    def _invalidate_opposing_active_signals(
+        self,
+        row: DecisionSignalRecord,
+        *,
+        reference_at: Optional[datetime],
+    ) -> None:
+        opposing_actions = self._opposing_actions(row.action)
+        if not opposing_actions:
+            return
+        old_rows = self.repo.list_active_by_stock_actions(
+            market=row.market,
+            stock_code=row.stock_code,
+            actions=sorted(opposing_actions),
+            exclude_signal_id=row.id,
+        )
+        for old_row in old_rows:
+            if not self._is_prior_signal(old_row, row, reference_at=reference_at):
+                continue
+            metadata_json = self._invalidation_metadata_json(old_row, invalidated_by=row)
+            updated = self.repo.update_status(
+                old_row.id,
+                status="invalidated",
+                metadata_json=metadata_json,
+                replace_metadata=True,
+            )
+            if updated is None:
+                logger.warning(
+                    "Decision signal disappeared before invalidation: signal_id=%s invalidated_by=%s",
+                    old_row.id,
+                    row.id,
+                )
+
+    @staticmethod
+    def _is_prior_signal(
+        candidate: DecisionSignalRecord,
+        current: DecisionSignalRecord,
+        *,
+        reference_at: Optional[datetime],
+    ) -> bool:
+        candidate_created_at = candidate.created_at
+        if candidate_created_at is not None and reference_at is not None:
+            candidate_created_at = to_utc_naive_datetime(candidate_created_at)
+            reference_at = to_utc_naive_datetime(reference_at)
+            if candidate_created_at != reference_at:
+                return candidate_created_at < reference_at
+
+        if candidate.id is not None and current.id is not None:
+            return candidate.id < current.id
+        return False
+
+    @staticmethod
+    def _opposing_actions(action: str) -> frozenset[str]:
+        if action in BULLISH_ACTIONS:
+            return DEFENSIVE_ACTIONS
+        if action in DEFENSIVE_ACTIONS:
+            return BULLISH_ACTIONS
+        return frozenset()
+
+    def _invalidation_metadata_json(
+        self,
+        row: DecisionSignalRecord,
+        *,
+        invalidated_by: DecisionSignalRecord,
+    ) -> Optional[str]:
+        metadata = self._metadata_for_invalidation(row)
+        metadata.update({
+            "invalidated_by_signal_id": invalidated_by.id,
+            "invalidated_reason": f"opposite_active_signal:{row.action}->{invalidated_by.action}",
+            "invalidated_at": utc_naive_now().isoformat(),
+            "previous_status": row.status,
+        })
+        return self._json_dumps(metadata)
+
+    @staticmethod
+    def _metadata_for_invalidation(row: DecisionSignalRecord) -> Dict[str, Any]:
+        if not row.metadata_json:
+            return {}
+        try:
+            value = json.loads(row.metadata_json)
+        except json.JSONDecodeError as exc:
+            logger.warning(
+                "Replacing invalid decision signal metadata during invalidation: id=%s error=%s",
+                row.id,
+                exc,
+            )
+            return {"metadata_replaced_due_to_invalid_json": True}
+        if isinstance(value, dict):
+            return dict(value)
+        return {"metadata_replaced_due_to_non_object": True}
 
     def _normalize_plan_quality(self, value: Any, *, fields: Dict[str, Any]) -> str:
         if value is not None:

tests/test_api_schema_pydantic.py

@@ -4,6 +4,7 @@ import json
 from pathlib import Path
 
 from api.app import create_app
+from api.v1.router import router as api_v1_router
 from api.v1.schemas.analysis import AnalyzeRequest, MarketReviewRequest
 from api.v1.schemas.common import RootResponse
 from api.v1.schemas.history import HistoryItem
@@ -107,3 +108,13 @@ def test_decision_signal_static_api_spec_matches_runtime_paths() -> None:
 
     status_schema = static_spec["components"]["schemas"]["DecisionSignalStatusUpdateRequest"]["properties"]["status"]
     assert status_schema["enum"] == ["active", "expired", "invalidated", "closed", "archived"]
+
+
+def test_v1_prefix_is_applied_at_app_mount_level() -> None:
+    assert api_v1_router.prefix == ""
+
+    runtime_paths = create_app().openapi()["paths"]
+    assert "/api/v1/history" in runtime_paths
+    assert "/api/v1/decision-signals" in runtime_paths
+    assert "/api/v1/history/" not in runtime_paths
+    assert "/api/v1/decision-signals/" not in runtime_paths

tests/test_decision_signal_api.py

@@ -173,6 +173,7 @@ def test_create_duplicate_list_detail_latest_and_status_update(client_and_db) ->
     signal_id = created["item"]["id"]
     assert created["item"]["stock_code"] == "600519"
     assert created["item"]["plan_quality"] == "partial"
+    assert created["item"]["expires_at"] is not None
 
     duplicate_resp = client.post(
         "/api/v1/decision-signals",
@@ -226,13 +227,12 @@ def test_create_duplicate_list_detail_latest_and_status_update(client_and_db) ->
     assert clear_metadata_resp.json()["status"] == "archived"
     assert clear_metadata_resp.json()["metadata"] is None
 
-    status_only_resp = client.patch(
+    terminal_reactivate_resp = client.patch(
         f"/api/v1/decision-signals/{signal_id}/status",
         json={"status": "active"},
     )
-    assert status_only_resp.status_code == 200, status_only_resp.text
-    assert status_only_resp.json()["status"] == "active"
-    assert status_only_resp.json()["metadata"] is None
+    assert terminal_reactivate_resp.status_code == 400, terminal_reactivate_resp.text
+    assert terminal_reactivate_resp.json()["error"] == "validation_error"
 
     invalid_status_resp = client.patch(
         f"/api/v1/decision-signals/{signal_id}/status",
@@ -245,6 +245,28 @@ def test_create_duplicate_list_detail_latest_and_status_update(client_and_db) ->
     assert missing_resp.status_code == 404
 
 
+def test_create_treats_null_lifecycle_fields_as_missing(client_and_db) -> None:
+    client, _db = client_and_db
+
+    response = client.post(
+        "/api/v1/decision-signals",
+        json=_payload(
+            source_report_id=3002,
+            trace_id="trace-null-lifecycle-api",
+            horizon=None,
+            expires_at=None,
+            market_phase="intraday",
+            metadata={"market_phase_summary": {"minutes_to_close": 25}},
+        ),
+    )
+
+    assert response.status_code == 200, response.text
+    item = response.json()["item"]
+    assert item["status"] == "active"
+    assert item["horizon"] == "intraday"
+    assert item["expires_at"] is not None
+
+
 def test_status_update_sanitizes_metadata_before_response_and_persistence(client_and_db) -> None:
     client, db = client_and_db
 
@@ -398,7 +420,6 @@ def test_patch_status_rejects_expired_signal_without_expires_at_extension(client
     assert created_resp.status_code == 200, created_resp.text
     item = created_resp.json()["item"]
     assert item["status"] == "expired"
-    assert item["expires_at"] is None
 
     reactivate_resp = client.patch(
         f"/api/v1/decision-signals/{item['id']}/status",
@@ -483,6 +504,43 @@ def test_create_refreshes_expired_same_source_when_future_expiry_is_supplied(cli
         assert row.trace_id == "trace-refresh-original"
 
 
+def test_create_invalidates_opposing_active_signal_and_latest_filters_it(client_and_db) -> None:
+    client, _db = client_and_db
+    buy_resp = client.post(
+        "/api/v1/decision-signals",
+        json=_payload(
+            source_report_id=31101,
+            trace_id="trace-api-opposing-buy",
+            action="buy",
+        ),
+    )
+    assert buy_resp.status_code == 200, buy_resp.text
+    buy = buy_resp.json()["item"]
+
+    sell_resp = client.post(
+        "/api/v1/decision-signals",
+        json=_payload(
+            source_report_id=31102,
+            trace_id="trace-api-opposing-sell",
+            action="sell",
+        ),
+    )
+    assert sell_resp.status_code == 200, sell_resp.text
+    sell = sell_resp.json()["item"]
+
+    old_resp = client.get(f"/api/v1/decision-signals/{buy['id']}")
+    assert old_resp.status_code == 200, old_resp.text
+    old = old_resp.json()
+    assert old["status"] == "invalidated"
+    assert old["metadata"]["invalidated_by_signal_id"] == sell["id"]
+
+    latest_resp = client.get("/api/v1/decision-signals/latest/600519", params={"limit": 5})
+    assert latest_resp.status_code == 200, latest_resp.text
+    latest = latest_resp.json()
+    assert latest["total"] == 1
+    assert latest["items"][0]["id"] == sell["id"]
+
+
 def test_create_does_not_refresh_expired_same_source_without_future_active_expiry(client_and_db) -> None:
     client, _db = client_and_db
     expired_resp = client.post(

tests/test_decision_signal_extractor.py

@@ -77,7 +77,12 @@ def test_build_payload_maps_report_context_and_price_plan() -> None:
     result.market_phase_summary = {"phase": "postmarket"}
     result.analysis_context_pack_overview = {"data_quality": {"overall_score": 55, "level": "fair"}}
     context_snapshot = {
-        "market_phase_summary": {"phase": "intraday"},
+        "market_phase_summary": {
+            "phase": "intraday",
+            "session_date": "2026-06-15",
+            "minutes_to_open": None,
+            "minutes_to_close": 120,
+        },
         "analysis_context_pack_overview": {
             "data_quality": {"overall_score": 91, "level": "good"},
         },
@@ -113,6 +118,11 @@ def test_build_payload_maps_report_context_and_price_plan() -> None:
     assert payload["risk_summary"] == ["跌破支撑需止损", "估值偏高"]
     assert payload["catalyst_summary"] == ["业绩超预期"]
     assert payload["metadata"]["report_confidence_level"] == "高"
+    assert payload["metadata"]["market_phase_summary"] == {
+        "phase": "intraday",
+        "session_date": "2026-06-15",
+        "minutes_to_close": 120,
+    }
 
 
 def test_build_payload_uses_result_fallbacks_and_optional_catalysts() -> None:
@@ -261,6 +271,8 @@ def test_extract_and_persist_reuses_service_dedup_and_sanitization(isolated_db)
     assert second["created"] is False
     assert first["item"]["reason"] == "趋势确认 token=[REDACTED]"
     assert first["item"]["plan_quality"] == "complete"
+    assert first["item"]["horizon"] == "intraday"
+    assert first["item"]["expires_at"] is not None
 
     listed = service.list_signals(source_report_id=901)
     assert listed["total"] == 1
@@ -289,6 +301,8 @@ def test_extract_and_persist_missing_price_plan_does_not_fabricate_fields(isolat
     assert created is not None
     item = created["item"]
     assert item["plan_quality"] == "minimal"
+    assert item["horizon"] == "3d"
+    assert item["expires_at"] is not None
     assert item["entry_low"] is None
     assert item["entry_high"] is None
     assert item["stop_loss"] is None

tests/test_decision_signal_repo.py

@@ -157,6 +157,142 @@ def test_create_if_absent_deduplicates_report_and_trace_keys(isolated_db) -> Non
     assert no_key_row2.id != no_key_row1.id
 
 
+def test_create_if_absent_relaxed_merge_only_fills_missing_default_dimensions(isolated_db) -> None:
+    repo = DecisionSignalRepository(isolated_db)
+
+    original = repo.create_if_absent(
+        _fields(
+            source_report_id=2401,
+            trace_id="trace-relaxed-original",
+            horizon=None,
+            market_phase=None,
+            reason="original reason",
+        )
+    )
+    merged = repo.create_if_absent(
+        _fields(
+            source_report_id=2401,
+            trace_id="trace-relaxed-new",
+            horizon="3d",
+            market_phase="intraday",
+            reason="new reason",
+        ),
+        allow_relaxed_horizon_fill=True,
+    )
+
+    assert original.created is True
+    assert merged.created is False
+    assert merged.refreshed is True
+    assert merged.duplicate is False
+    assert merged.row.id == original.row.id
+    assert merged.row.horizon == "3d"
+    assert merged.row.market_phase == "intraday"
+    assert merged.row.reason == "original reason"
+
+    duplicate = repo.create_if_absent(
+        _fields(
+            source_report_id=2401,
+            trace_id="trace-relaxed-duplicate",
+            horizon="3d",
+            market_phase="intraday",
+        ),
+        allow_relaxed_horizon_fill=True,
+    )
+    assert duplicate.created is False
+    assert duplicate.refreshed is False
+    assert duplicate.duplicate is True
+    assert duplicate.row.id == original.row.id
+
+    explicit_horizon = repo.create_if_absent(
+        _fields(
+            source_report_id=2402,
+            trace_id="trace-explicit-horizon-original",
+            horizon=None,
+            market_phase=None,
+        )
+    )
+    explicit_horizon_new = repo.create_if_absent(
+        _fields(
+            source_report_id=2402,
+            trace_id="trace-explicit-horizon-new",
+            horizon="swing",
+            market_phase="intraday",
+        ),
+        allow_relaxed_horizon_fill=False,
+    )
+    assert explicit_horizon.created is True
+    assert explicit_horizon_new.created is True
+    assert explicit_horizon_new.row.id != explicit_horizon.row.id
+
+    different_phase = repo.create_if_absent(
+        _fields(
+            source_report_id=2403,
+            trace_id="trace-different-phase-original",
+            horizon=None,
+            market_phase="postmarket",
+        )
+    )
+    different_phase_new = repo.create_if_absent(
+        _fields(
+            source_report_id=2403,
+            trace_id="trace-different-phase-new",
+            horizon="3d",
+            market_phase="intraday",
+        ),
+        allow_relaxed_horizon_fill=True,
+    )
+    assert different_phase.created is True
+    assert different_phase_new.created is True
+    assert different_phase_new.row.id != different_phase.row.id
+
+
+def test_create_if_absent_relaxed_merge_skips_terminal_candidates(isolated_db) -> None:
+    repo = DecisionSignalRepository(isolated_db)
+    closed = repo.create(
+        _fields(
+            source_report_id=2404,
+            trace_id="trace-relaxed-closed",
+            horizon=None,
+            market_phase=None,
+            status="closed",
+            reason="closed reason",
+        )
+    )
+    active = repo.create(
+        _fields(
+            source_report_id=2404,
+            trace_id="trace-relaxed-active",
+            horizon=None,
+            market_phase=None,
+            reason="active reason",
+        )
+    )
+
+    merged = repo.create_if_absent(
+        _fields(
+            source_report_id=2404,
+            trace_id="trace-relaxed-new-active",
+            horizon="3d",
+            market_phase="intraday",
+            reason="new reason",
+        ),
+        allow_relaxed_horizon_fill=True,
+    )
+
+    assert merged.created is False
+    assert merged.refreshed is True
+    assert merged.row.id == active.id
+    assert merged.row.horizon == "3d"
+    assert merged.row.market_phase == "intraday"
+    assert merged.row.reason == "active reason"
+
+    closed_after = repo.get(closed.id)
+    assert closed_after is not None
+    assert closed_after.status == "closed"
+    assert closed_after.horizon is None
+    assert closed_after.market_phase is None
+
+
 def test_list_latest_status_update_and_lazy_expire(isolated_db) -> None:
     repo = DecisionSignalRepository(isolated_db)
     old_row = repo.create(_fields(source_report_id=2001, trace_id="trace-2001", action="watch"))

tests/test_decision_signal_service.py

@@ -6,14 +6,17 @@ from __future__ import annotations
 import os
 import subprocess
 import sys
+from datetime import datetime, timedelta
 from math import inf, nan
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
 from src.config import Config
+from src.repositories.decision_signal_repo import DecisionSignalCreateResult
 from src.services.decision_signal_service import DecisionSignalService, DecisionSignalStorageError
-from src.storage import DatabaseManager, DecisionSignalRecord
+from src.storage import DatabaseManager, DecisionSignalRecord, utc_naive_now
 from src.utils.sanitize import sanitize_decision_signal_text, sanitize_diagnostic_text
 
 
@@ -98,6 +101,114 @@ def test_service_normalizes_fields_and_partial_plan_quality(isolated_db) -> None
     assert item["plan_quality"] == "partial"
 
 
+def test_service_defaults_lifecycle_and_preserves_explicit_values(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    intraday_payload = _payload(
+        source_report_id=151,
+        trace_id="trace-lifecycle-intraday",
+        market_phase="intraday",
+        metadata={"market_phase_summary": {"minutes_to_close": 45}},
+    )
+    intraday_payload.pop("horizon")
+    before_intraday = utc_naive_now()
+    intraday = service.create_signal(intraday_payload)["item"]
+    intraday_expiry = datetime.fromisoformat(intraday["expires_at"])
+    assert intraday["horizon"] == "intraday"
+    assert before_intraday + timedelta(minutes=44) <= intraday_expiry
+    assert intraday_expiry <= utc_naive_now() + timedelta(minutes=46)
+
+    opening_payload = _payload(
+        source_report_id=157,
+        trace_id="trace-lifecycle-opening",
+        market_phase="premarket",
+        metadata={"market_phase_summary": {"minutes_to_open": 10}},
+    )
+    opening_payload.pop("horizon")
+    before_opening = utc_naive_now()
+    opening = service.create_signal(opening_payload)["item"]
+    opening_expiry = datetime.fromisoformat(opening["expires_at"])
+    assert opening["horizon"] == "intraday"
+    assert before_opening + timedelta(hours=4, minutes=9) <= opening_expiry
+    assert opening_expiry <= utc_naive_now() + timedelta(hours=4, minutes=11)
+
+    hk_alert_payload = _payload(
+        source_report_id=152,
+        trace_id="trace-lifecycle-hk-alert",
+        stock_code="00700",
+        stock_name="Tencent",
+        market="hk",
+        action="alert",
+    )
+    hk_alert_payload.pop("horizon")
+    hk_alert_payload.pop("market_phase")
+    before_alert = utc_naive_now()
+    hk_alert = service.create_signal(hk_alert_payload)["item"]
+    hk_alert_expiry = datetime.fromisoformat(hk_alert["expires_at"])
+    assert hk_alert["horizon"] == "intraday"
+    assert before_alert + timedelta(hours=5, minutes=29) <= hk_alert_expiry
+    assert hk_alert_expiry <= utc_naive_now() + timedelta(hours=5, minutes=31)
+
+    postmarket_payload = _payload(
+        source_report_id=153,
+        trace_id="trace-lifecycle-postmarket",
+        market_phase="postmarket",
+    )
+    postmarket_payload.pop("horizon")
+    before_postmarket = utc_naive_now()
+    postmarket = service.create_signal(postmarket_payload)["item"]
+    postmarket_expiry = datetime.fromisoformat(postmarket["expires_at"])
+    assert postmarket["horizon"] == "3d"
+    assert before_postmarket + timedelta(days=3, seconds=-1) <= postmarket_expiry
+    assert postmarket_expiry <= utc_naive_now() + timedelta(days=3, seconds=1)
+
+    null_lifecycle_payload = _payload(
+        source_report_id=158,
+        trace_id="trace-lifecycle-null-values",
+        horizon=None,
+        expires_at=None,
+        market_phase="intraday",
+        metadata={"market_phase_summary": {"minutes_to_close": 30}},
+    )
+    before_null_lifecycle = utc_naive_now()
+    null_lifecycle = service.create_signal(null_lifecycle_payload)["item"]
+    null_lifecycle_expiry = datetime.fromisoformat(null_lifecycle["expires_at"])
+    assert null_lifecycle["horizon"] == "intraday"
+    assert before_null_lifecycle + timedelta(minutes=29) <= null_lifecycle_expiry
+    assert null_lifecycle_expiry <= utc_naive_now() + timedelta(minutes=31)
+
+    swing = service.create_signal(
+        _payload(
+            source_report_id=154,
+            trace_id="trace-lifecycle-swing",
+            horizon="swing",
+        )
+    )["item"]
+    assert swing["horizon"] == "swing"
+    assert swing["expires_at"] is None
+
+    explicit_expires_at = "2099-01-01T00:00:00Z"
+    explicit = service.create_signal(
+        _payload(
+            source_report_id=155,
+            trace_id="trace-lifecycle-explicit",
+            horizon="1d",
+            expires_at=explicit_expires_at,
+        )
+    )["item"]
+    assert explicit["horizon"] == "1d"
+    assert explicit["expires_at"] == "2099-01-01T00:00:00"
+
+    past = service.create_signal(
+        _payload(
+            source_report_id=156,
+            trace_id="trace-lifecycle-past",
+            expires_at=(utc_naive_now() - timedelta(minutes=1)).isoformat(),
+        )
+    )["item"]
+    assert past["status"] == "expired"
+
+
 def test_service_plan_quality_slots_and_explicit_override(isolated_db) -> None:
     service = DecisionSignalService(db_manager=isolated_db)
 
@@ -475,3 +586,238 @@ def test_service_raises_on_corrupt_persisted_json(isolated_db) -> None:
 
     with pytest.raises(DecisionSignalStorageError, match="invalid persisted JSON"):
         service.get_signal(signal_id)
+
+
+@pytest.mark.parametrize("terminal_status", ["expired", "invalidated", "closed", "archived"])
+def test_service_rejects_terminal_status_reactivation(isolated_db, terminal_status) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    created = service.create_signal(
+        _payload(source_report_id=360, trace_id=f"trace-terminal-{terminal_status}")
+    )
+    signal_id = created["item"]["id"]
+
+    service.update_status(signal_id, status=terminal_status)
+
+    with pytest.raises(ValueError, match="terminal decision signal"):
+        service.update_status(signal_id, status="active")
+
+
+def test_service_invalidates_opposing_active_signals(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    old_buy = service.create_signal(
+        _payload(
+            source_report_id=371,
+            trace_id="trace-opposing-buy",
+            action="buy",
+            metadata={"task_id": "old-buy"},
+        )
+    )["item"]
+
+    new_sell = service.create_signal(
+        _payload(
+            source_report_id=372,
+            trace_id="trace-opposing-sell",
+            action="sell",
+        )
+    )["item"]
+
+    old_after = service.get_signal(old_buy["id"])
+    assert new_sell["status"] == "active"
+    assert old_after["status"] == "invalidated"
+    assert old_after["metadata"]["task_id"] == "old-buy"
+    assert old_after["metadata"]["invalidated_by_signal_id"] == new_sell["id"]
+    assert old_after["metadata"]["invalidated_reason"] == "opposite_active_signal:buy->sell"
+    assert old_after["metadata"]["previous_status"] == "active"
+
+    latest = service.get_latest_active(stock_code="600519", limit=5)
+    assert [item["id"] for item in latest["items"]] == [new_sell["id"]]
+
+
+def test_service_expired_refresh_invalidates_later_opposing_active_signal(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    buy_payload = _payload(source_report_id=376, trace_id="trace-refresh-buy", action="buy")
+    old_buy = service.create_signal(buy_payload)["item"]
+    service.update_status(old_buy["id"], status="expired")
+
+    active_sell = service.create_signal(
+        _payload(source_report_id=377, trace_id="trace-refresh-sell", action="sell")
+    )["item"]
+    assert service.get_signal(active_sell["id"])["status"] == "active"
+
+    refreshed = service.create_signal(
+        {
+            **buy_payload,
+            "expires_at": (utc_naive_now() + timedelta(days=1)).isoformat(),
+        }
+    )
+
+    assert refreshed["created"] is False
+    assert refreshed["item"]["id"] == old_buy["id"]
+    assert refreshed["item"]["status"] == "active"
+    sell_after = service.get_signal(active_sell["id"])
+    assert sell_after["status"] == "invalidated"
+    assert sell_after["metadata"]["invalidated_by_signal_id"] == old_buy["id"]
+    latest = service.get_latest_active(stock_code="600519", limit=5)
+    assert [item["id"] for item in latest["items"]] == [old_buy["id"]]
+
+
+def test_service_does_not_invalidate_neutral_or_terminal_signals(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    old_buy = service.create_signal(
+        _payload(source_report_id=381, trace_id="trace-neutral-buy", action="buy")
+    )["item"]
+
+    hold = service.create_signal(
+        _payload(source_report_id=382, trace_id="trace-neutral-hold", action="hold")
+    )["item"]
+
+    assert hold["status"] == "active"
+    assert service.get_signal(old_buy["id"])["status"] == "active"
+
+    service.update_status(old_buy["id"], status="closed")
+    service.create_signal(
+        _payload(source_report_id=383, trace_id="trace-terminal-sell", action="sell")
+    )
+    assert service.get_signal(old_buy["id"])["status"] == "closed"
+
+
+def test_service_replaces_corrupt_metadata_during_invalidation(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    old_buy = service.create_signal(
+        _payload(source_report_id=391, trace_id="trace-corrupt-metadata-buy", action="buy")
+    )["item"]
+
+    with isolated_db.get_session() as session:
+        row = session.query(DecisionSignalRecord).filter_by(id=old_buy["id"]).one()
+        row.metadata_json = "{not valid json"
+        session.commit()
+
+    new_sell = service.create_signal(
+        _payload(source_report_id=392, trace_id="trace-corrupt-metadata-sell", action="sell")
+    )["item"]
+
+    old_after = service.get_signal(old_buy["id"])
+    assert old_after["status"] == "invalidated"
+    assert old_after["metadata"]["metadata_replaced_due_to_invalid_json"] is True
+    assert old_after["metadata"]["invalidated_by_signal_id"] == new_sell["id"]
+
+
+def test_service_replaces_non_object_metadata_during_invalidation(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    old_buy = service.create_signal(
+        _payload(source_report_id=393, trace_id="trace-non-object-metadata-buy", action="buy")
+    )["item"]
+
+    with isolated_db.get_session() as session:
+        row = session.query(DecisionSignalRecord).filter_by(id=old_buy["id"]).one()
+        row.metadata_json = '["legacy"]'
+        session.commit()
+
+    new_sell = service.create_signal(
+        _payload(source_report_id=394, trace_id="trace-non-object-metadata-sell", action="sell")
+    )["item"]
+
+    old_after = service.get_signal(old_buy["id"])
+    assert old_after["status"] == "invalidated"
+    assert old_after["metadata"]["metadata_replaced_due_to_non_object"] is True
+    assert old_after["metadata"]["invalidated_by_signal_id"] == new_sell["id"]
+
+
+def test_service_duplicate_retry_repairs_failed_invalidation(isolated_db, monkeypatch) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    old_buy = service.create_signal(
+        _payload(source_report_id=392, trace_id="trace-repair-buy", action="buy")
+    )["item"]
+    sell_payload = _payload(source_report_id=393, trace_id="trace-repair-sell", action="sell")
+    original_update_status = service.repo.update_status
+
+    def fail_once(*_args, **_kwargs):
+        raise RuntimeError("invalidation write failed")
+
+    monkeypatch.setattr(service.repo, "update_status", fail_once)
+    with pytest.raises(RuntimeError, match="invalidation write failed"):
+        service.create_signal(sell_payload)
+
+    assert service.get_signal(old_buy["id"])["status"] == "active"
+
+    monkeypatch.setattr(service.repo, "update_status", original_update_status)
+    retried = service.create_signal(sell_payload)
+
+    assert retried["created"] is False
+    assert retried["item"]["status"] == "active"
+    old_after = service.get_signal(old_buy["id"])
+    assert old_after["status"] == "invalidated"
+    assert old_after["metadata"]["invalidated_by_signal_id"] == retried["item"]["id"]
+
+
+def test_service_duplicate_old_signal_does_not_invalidate_newer_opposing_signal(isolated_db, monkeypatch) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    buy_payload = _payload(source_report_id=395, trace_id="trace-old-replay-buy", action="buy")
+    old_buy = service.create_signal(buy_payload)["item"]
+
+    monkeypatch.setattr(service, "_invalidate_opposing_active_signals", lambda *_args, **_kwargs: None)
+    new_sell = service.create_signal(
+        _payload(source_report_id=396, trace_id="trace-old-replay-sell", action="sell")
+    )["item"]
+    monkeypatch.undo()
+
+    replayed_buy = service.create_signal(buy_payload)
+
+    assert replayed_buy["created"] is False
+    assert replayed_buy["item"]["id"] == old_buy["id"]
+    assert service.get_signal(new_sell["id"])["status"] == "active"
+    assert service.get_signal(old_buy["id"])["status"] == "active"
+
+
+def test_service_relaxed_active_fill_does_not_invalidate_newer_opposing_signal(isolated_db, monkeypatch) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    buy_payload = _payload(source_report_id=397, trace_id="trace-relaxed-fill-buy", action="buy")
+    old_buy = service.create_signal(buy_payload)["item"]
+
+    with isolated_db.get_session() as session:
+        row = session.query(DecisionSignalRecord).filter_by(id=old_buy["id"]).one()
+        row.horizon = None
+        row.market_phase = None
+        session.commit()
+
+    monkeypatch.setattr(service, "_invalidate_opposing_active_signals", lambda *_args, **_kwargs: None)
+    new_sell = service.create_signal(
+        _payload(source_report_id=398, trace_id="trace-relaxed-fill-sell", action="sell")
+    )["item"]
+    monkeypatch.undo()
+
+    relaxed_payload = dict(buy_payload)
+    relaxed_payload.pop("horizon")
+    replayed_buy = service.create_signal(relaxed_payload)
+
+    assert replayed_buy["created"] is False
+    assert replayed_buy["item"]["id"] == old_buy["id"]
+    assert replayed_buy["item"]["horizon"] == "intraday"
+    assert replayed_buy["item"]["market_phase"] == "intraday"
+    assert service.get_signal(new_sell["id"])["status"] == "active"
+    assert service.get_signal(old_buy["id"])["status"] == "active"
+
+
+def test_service_propagates_unexpected_invalidation_failures(isolated_db) -> None:
+    class FailingInvalidationRepo:
+        def create_if_absent(self, fields, *, allow_relaxed_horizon_fill=False):
+            row = SimpleNamespace(
+                id=1,
+                status="active",
+                action=fields["action"],
+                market=fields["market"],
+                stock_code=fields["stock_code"],
+            )
+            return DecisionSignalCreateResult(
+                row=row,
+                created=True,
+                invalidation_reference_at=utc_naive_now(),
+            )
+
+        def list_active_by_stock_actions(self, **_kwargs):
+            raise RuntimeError("invalidation write failed")
+
+    service = DecisionSignalService(repo=FailingInvalidationRepo(), db_manager=isolated_db)
+
+    with pytest.raises(RuntimeError, match="invalidation write failed"):
+        service.create_signal(_payload(source_report_id=392, trace_id="trace-invalidation-failure"))