From e6830471adae797a2fc489e3fc2df49e251aabdc Mon Sep 17 00:00:00 2001 From: lipluscodex <268560960+lipluscodex@users.noreply.github.com> Date: Fri, 21 Aug 2026 21:14:06 +0900 Subject: [PATCH 1/2] feat: add outcome-driven feedback deactivation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit soft-start contributionのcredited deltaと同一contributionのsibling normalization deltaを一体で永続化し、帰属可能なnegative outcomeだけを可逆に不活性化します。correctedと olled_backはbaseline floorを保ってexact reversalし、supersededはedgeを削除せずdormant化し、独立した後続confirmedで再活性化します。 default-offのcore/MCP/config provenance、mechanics test、fresh private snapshot向けresult-free protocol・runner・verifier・preflightを追加します。このcommitではdevelopment/holdout outputを生成せず、観測はsquash merge後のsuccessor Issueへ分離します。 --- README.md | 2 + docs/Decision-Structure.md | 1 + docs/optional-mcp-interface.md | 29 +- docs/outcome-driven-feedback-deactivation.md | 60 ++ docs/requirements.md | 3 + src/neuron_graph_rag/__init__.py | 6 + src/neuron_graph_rag/config_provenance.py | 3 + src/neuron_graph_rag/evidence_feedback.py | 27 + src/neuron_graph_rag/feedback.py | 75 ++ src/neuron_graph_rag/models.py | 36 + ...utcome_feedback_deactivation_evaluation.py | 641 ++++++++++++++++++ src/neuron_graph_rag/storage.py | 487 ++++++++++++- src/neuron_graph_rag_mcp/server.py | 134 +++- ...come_feedback_deactivation_v1.fixture.json | 130 ++++ ...outcome_feedback_deactivation_v1.gate.json | 37 + ...ome_feedback_deactivation_v1.manifest.json | 72 ++ ...ack_deactivation_v1.result-free-audit.json | 13 + ...eedback_deactivation_v1.result-schema.json | 21 + ...ome_feedback_deactivation_v1.schedule.json | 52 ++ tests/test_mcp_adapter.py | 91 +++ tests/test_outcome_feedback_deactivation.py | 326 +++++++++ ...utcome_feedback_deactivation_evaluation.py | 157 +++++ ...utcome_feedback_deactivation_evaluation.py | 47 ++ 23 files changed, 2439 insertions(+), 11 deletions(-) create mode 100644 docs/outcome-driven-feedback-deactivation.md create mode 100644 src/neuron_graph_rag/outcome_feedback_deactivation_evaluation.py create mode 100644 tests/fixtures/outcome_feedback_deactivation_v1.fixture.json create mode 100644 tests/fixtures/outcome_feedback_deactivation_v1.gate.json create mode 100644 tests/fixtures/outcome_feedback_deactivation_v1.manifest.json create mode 100644 tests/fixtures/outcome_feedback_deactivation_v1.result-free-audit.json create mode 100644 tests/fixtures/outcome_feedback_deactivation_v1.result-schema.json create mode 100644 tests/fixtures/outcome_feedback_deactivation_v1.schedule.json create mode 100644 tests/test_outcome_feedback_deactivation.py create mode 100644 tests/test_outcome_feedback_deactivation_evaluation.py create mode 100644 tools/run_outcome_feedback_deactivation_evaluation.py diff --git a/README.md b/README.md index 8c3d940..acce88e 100644 --- a/README.md +++ b/README.md @@ -298,6 +298,8 @@ Confirmed-outcome feedback reinforcementは、同じ candidate class の `confir Soft-start feedback reinforcementは、`soft_start_feedback_reinforcement=True`、`soft_start_feedback_ratio`、`confirmation_decay_ratio` を明示する別の default-off candidate です。credited relation edge の最初の新規 `used` は通常 bounded update の一部だけを provisional に適用し、最初の独立 `confirmed` が残りを補完します。後続 confirmation は既存 geometric decay に従い、same-source sibling normalization は `used` ではなく各 confirmation の actual delta だけへ適用します。confirmed-only candidate、hard quorum と同時には有効化できず、既存 q1/s0、q3/s1、confirmed-only、凍結評価 artifact と default fingerprint は変更しません。詳細は[Confirmed-outcome feedback reinforcement](docs/confirmed-outcome-feedback-reinforcement.md#soft-start-successor-candidate)を参照してください。 +Outcome-driven feedback deactivationは、soft-startへ `outcome_driven_feedback_deactivation=True` を追加した時だけ有効になるdefault-off candidateです。保存済みtrace / credited pathへ帰属できる`corrected`と`rolled_back`は、credited加算と同じcontributionのsame-source sibling normalizationを一体で正確に戻します。`superseded`はedgeと証拠を削除せずdormantにし、同じ保存済みpathの後続`confirmed`で再活性化します。lexical、zero-hop、別source、uncredited failure、基礎weight以下へのpunitive updateは変更しません。mechanicsとresult-free snapshot protocolだけを先にfreezeし、観測はmerge後のsuccessor Issueへ分離します。詳細は[Outcome-driven feedback deactivation](docs/outcome-driven-feedback-deactivation.md)を参照してください。 + local database上の比較は、[Soft-start snapshot evaluation](docs/soft-start-snapshot-evaluation.md)でtransaction-consistent private snapshot、固定4 arm、result-free manifest、one-time development、conditional holdout、privacy / mutation / rank hard gateとして分離しています。評価はsnapshotのfresh cloneだけを変更し、source database、live config、snapshot本体、private本文をpublic artifactへ含めません。支持結果も固定local snapshot上のcutover候補に限定し、このissue内でconfigやdefaultを変更しません。 v1 developmentはsnapshot既存evidenceによりq3/s1がfresh `used_2`で発火し、空baseline由来の`used_3`を固定したboundary gateだけが不合格となりました。この不支持resultを変更せず、[Baseline-aware soft-start snapshot evaluation](docs/baseline-aware-soft-start-snapshot-evaluation.md)はfresh snapshotと別caseを使い、initial evidence stateからq3 first mutationを導出します。baseline不一致やcapacity不足ではregistered resultを作りません。v2はregistered output不在のfreeze-only PRとして固定し、squash merge後の別Issueでdevelopmentを一度だけ実行して、全gate通過時だけholdoutを一度開きます。 diff --git a/docs/Decision-Structure.md b/docs/Decision-Structure.md index 67f9902..c708cd6 100644 --- a/docs/Decision-Structure.md +++ b/docs/Decision-Structure.md @@ -14,6 +14,7 @@ | [ceiling-aware-feedback-adaptation-gate](https://github.com/Liplus-Project/neuron-graph-rag/wiki/ceiling-aware-feedback-adaptation-gate) | active | 新規 feedback-adaptation experiment は baseline relation MRR が 1.0 未満なら strict improvement、1.0 なら全 safety gate を満たす non-regression を要求する。ceiling pass は default や一般化を意味しない。 | | [evidence-gated-local-feedback-reinforcement](https://github.com/Liplus-Project/neuron-graph-rag/wiki/evidence-gated-local-feedback-reinforcement) | active | relation edge ごとの独立 success trace が固定 quorum に達するまで serving weight を変えず、到達後は既存 bounded reinforcement と same-source sibling normalization を一回ずつ適用する。candidate は default-preserving であり、採用値と一般化を主張しない。 | | [confirmed-outcome-feedback-reinforcement](https://github.com/Liplus-Project/neuron-graph-rag/wiki/confirmed-outcome-feedback-reinforcement) | superseded | confirmed-only candidate の mechanics と証拠は保持するが、利用直後の小さな適応を残す [soft-start-feedback-reinforcement](https://github.com/Liplus-Project/neuron-graph-rag/wiki/soft-start-feedback-reinforcement) が後続比較の現在候補としてこの判断を supersede する。 | +| [outcome-driven-feedback-deactivation](https://github.com/Liplus-Project/neuron-graph-rag/wiki/outcome-driven-feedback-deactivation) | evaluating | 因果帰属できるnegative outcomeだけがsoft-start contributionと同一単位のsibling normalizationを正確に戻し、superseded edgeは削除せずdormant / reactivationで扱う。mechanicsとresult-free freezeを先に固定し、観測と採用判断はsuccessor Issueへ分離する。 | | [frozen-output-round-trip-integrity](https://github.com/Liplus-Project/neuron-graph-rag/wiki/frozen-output-round-trip-integrity) | active | result-free evaluation は canonical gate ID array を唯一の順序正本とし、freeze 前に登録外 placeholder と temporary output で実 writer から実 verifier への非アルファベット順 round-trip を証明する。observed result は exclusive creation し、array の完全性、順序、重複なし、全 gate pass を検証する。 | | [longitudinal-feedback-adaptation](https://github.com/Liplus-Project/neuron-graph-rag/wiki/longitudinal-feedback-adaptation) | active | longitudinal feedback-adaptation は、repository-native controlled corpus v3 の相互に独立した cluster と、その source 文書に明記した 0、1、3、10 credit ceiling を用いる。corpus phase は query、gold、schedule、runner、gate、manifest、result、既定値を定義・変更しない。 | | [single-corpus-real-feedback-validation](https://github.com/Liplus-Project/neuron-graph-rag/wiki/single-corpus-real-feedback-validation) | superseded | [repository-native-controlled-corpus](https://github.com/Liplus-Project/neuron-graph-rag/wiki/repository-native-controlled-corpus) がこの node を supersede する。以後の evaluation は、NGR repository に公開する固定 SHA の controlled corpus を source とし、D1 single-corpus experiment は capacity が増えるまで waiting とする。 | diff --git a/docs/optional-mcp-interface.md b/docs/optional-mcp-interface.md index 35b68f4..f3897b0 100644 --- a/docs/optional-mcp-interface.md +++ b/docs/optional-mcp-interface.md @@ -8,7 +8,7 @@ - MCP SDK が NGR core の必須依存である - 認証方式、transport、公開 endpoint、remote deployment が決定済みである -- delayed outcome が現在の edge weight を自動的に減算または巻き戻す +- delayed outcome が既定 policy で現在の edge weight を自動的に減算または巻き戻す `src/neuron_graph_rag_mcp/` の optional adapter がこの契約を local stdio transport で実装する。`pip install -e '.[mcp]'` で追加依存を導入し、`neuron-graph-rag-mcp --database ` で起動する。NGR core は引き続き Python 標準ライブラリだけで動作する。 @@ -52,6 +52,19 @@ neuron-graph-rag-mcp \ soft-start ratio と confirmation decay は有限の `0 < value < 1` とし、confirmed-only flag、hard evidence quorum と同時には有効化できない。MCP `search` は relation trace と active soft-start field を含む effective-config provenance を返す。最初の新規 `used` は通常 bounded update の provisional fraction を core と同じ receipt 形で返し、最初の独立 `confirmed` は残り、後続 confirmation は geometric decay を返す。`used` は sibling を変更せず、confirmation の actual delta だけが sibling normalization の対象になる。詳細は [Confirmed-outcome feedback reinforcement の soft-start 節](confirmed-outcome-feedback-reinforcement.md#soft-start-successor-candidate) を正本とする。 +outcome-driven deactivation candidate は上のsoft-start三値に次のflagを追加したprocessだけで有効になる。 + +```bash +neuron-graph-rag-mcp \ + --database /absolute/path/to/knowledge.db \ + --soft-start-feedback-reinforcement \ + --soft-start-feedback-ratio 0.25 \ + --confirmation-decay-ratio 0.5 \ + --outcome-driven-feedback-deactivation +``` + +帰属可能な`corrected` / `rolled_back`はcredited deltaと同一contributionのsibling normalization deltaを一体で逆適用し、`superseded`はedgeをdormantにする。後続`confirmed`は保存済みpathを再活性化する。MCP receiptは`deactivation_applied`、`reversed_contributions`、`dormancy_changes`、`reactivated_edges`をcoreと同じ形で返す。既定時のaudit-only動作とdescriptionは変更しない。詳細は[Outcome-driven feedback deactivation](outcome-driven-feedback-deactivation.md)を正本とする。 + ## 2. Protocol envelope tool 名は `search`、`record_source_use`、`record_outcome` とする。すべての input と成功 output は JSON Schema で宣言し、未知 field を受け付けない。 @@ -366,7 +379,7 @@ core domain API は取得済みでない node を stage 更新前に拒否し、 ### 7.1 Meaning -source を利用した判断や artifact に後から判明した結果を、即時 source-use とは別軸で記録する。既定 policy では評価用の履歴であり、edge weight を変更しない。明示 confirmed-only candidate では `confirmed` だけが保存済み relation credited path の diminishing reinforcement を発火できる。soft-start candidate では最初の `confirmed` が通常 update の残り、後続 `confirmed` が geometric decay を発火できる。 +source を利用した判断や artifact に後から判明した結果を、即時 source-use とは別軸で記録する。既定 policy では評価用の履歴であり、edge weight を変更しない。明示 confirmed-only candidate では `confirmed` だけが保存済み relation credited path の diminishing reinforcement を発火できる。soft-start candidate では最初の `confirmed` が通常 update の残り、後続 `confirmed` が geometric decay を発火できる。outcome-driven deactivation candidateでは、帰属可能なnegative outcomeだけが保存済みcontributionを可逆に不活性化できる。 ### 7.2 Normative model-facing description @@ -376,6 +389,12 @@ source を利用した判断や artifact に後から判明した結果を、即 Record a delayed outcome for sources that were already marked used, such as confirmed, corrected, rolled_back, or superseded. In v1, delayed outcomes are audit and evaluation records only: they do not add, subtract, undo, or otherwise change graph weights. Do not use this tool instead of record_source_use for immediate source-use feedback. If the trace handle has expired or does not exist, this tool returns unknown_trace. ``` +outcome-driven deactivation candidateを明示したprocessは、代わりに次をexact `description`とする。 + +```text +Record a delayed outcome for sources already marked used. This server uses the outcome-driven deactivation candidate: confirmed follows the soft-start schedule; causally attributed corrected and rolled_back outcomes exactly reverse each active credited contribution together with its same-source sibling normalization mutations. Superseded makes the saved relation path dormant without deleting evidence, and a later confirmed outcome on that saved path reactivates it. Unattributed, duplicate, lexical, and zero-hop outcomes remain non-mutating. +``` + ### 7.3 Outcome enum | outcome | 意味 | @@ -385,7 +404,7 @@ Record a delayed outcome for sources that were already marked used, such as conf | `rolled_back` | 判断または artifact が撤回、revert、rollback された | | `superseded` | 誤りと断定せず、新しい前提または判断に置き換えられた | -`corrected` と `rolled_back` を即時の負の reinforcement に変換しない。query、index、source selection、source 自体、実装のどこに原因があるかを一件の outcome だけで判別できないためである。既定 policy の `confirmed` も `used` の reinforcement を重複加算しない。soft-start の `confirmed` は provisional と合算して通常 update 一回を超えない remainder だけを最初に加算する。 +既定 policy では`corrected` と `rolled_back` を即時の負の reinforcement に変換しない。query、index、source selection、source 自体、実装のどこに原因があるかを一件の outcome だけで判別できないためである。明示deactivation candidateだけが保存済みtrace / credited pathへ帰属できるsoft-start contributionをexact reversalする。既定 policy の `confirmed` も `used` の reinforcement を重複加算しない。soft-start の `confirmed` は provisional と合算して通常 update 一回を超えない remainder だけを最初に加算する。 ### 7.4 Input @@ -430,11 +449,11 @@ Record a delayed outcome for sources that were already marked used, such as conf } ``` -既定 policy の `reinforcement_applied` は常に `false` とする。confirmed-only と soft-start candidate では新しい独立 edge confirmation を保存した時だけ `true` とし、`confirmations` に count、multiplier、actual delta、old/new weight、`credited_paths` に保存済み relation steps、`normalized_sibling_edges` に局所変更を返す。duplicate trace と idempotency replay は count と weight を重複変更しない。`corrected`、`rolled_back`、`superseded` は candidate 有効時も `false` のままである。 +既定 policy の `reinforcement_applied` は常に `false` とする。confirmed-only と soft-start candidate では新しい独立 edge confirmation を保存した時だけ `true` とし、`confirmations` に count、multiplier、actual delta、old/new weight、`credited_paths` に保存済み relation steps、`normalized_sibling_edges` に局所変更を返す。duplicate trace と idempotency replay は count と weight を重複変更しない。negative outcomeでは`reinforcement_applied`を`false`のまま保ち、明示deactivation candidateの変更有無は`deactivation_applied`、exact inverseは`reversed_contributions[].mutations`、dormancy遷移は`dormancy_changes`、再活性化は`reactivated_edges`に返す。 ### 7.6 Core mapping -transport-neutral な `FeedbackLedger.record_outcome` は既定 policy では outcome ledger にだけ保存する。confirmed-only と soft-start candidate では `record_success` を再利用せず、outcome、confirmation count、edge/sibling update、receipt を candidate 専用の一つの storage transaction へ渡す。 +transport-neutral な `FeedbackLedger.record_outcome` は既定 policy では outcome ledger にだけ保存する。confirmed-only と soft-start candidate では `record_success` を再利用せず、outcome、confirmation count、edge/sibling update、receipt を candidate 専用の一つの storage transaction へ渡す。deactivation candidateもoutcome、credited/sibling inverse、contribution state、dormancy、receiptを一つのtransactionへ渡す。 ## 8. Failure contract diff --git a/docs/outcome-driven-feedback-deactivation.md b/docs/outcome-driven-feedback-deactivation.md new file mode 100644 index 0000000..73852e0 --- /dev/null +++ b/docs/outcome-driven-feedback-deactivation.md @@ -0,0 +1,60 @@ +# Outcome-driven feedback deactivation + +## 目的と有効化境界 + +この文書は、soft-start が加算した relation feedback を、保存済み trace と credited path へ因果帰属できる negative delayed outcome に限って可逆に不活性化する candidate の正本である。candidate は `neuron_graph_rag.evidence_feedback.EngineConfig` の `outcome_driven_feedback_deactivation=True` で明示的に有効化し、`soft_start_feedback_reinforcement=True`、`soft_start_feedback_ratio`、`confirmation_decay_ratio` を同時に必要とする。 + +既定値は `False` である。無効時は `corrected`、`rolled_back`、`superseded` を従来どおり audit-only で保存し、既存 config fingerprint、library default、MCP default、local serving config を変更しない。candidate は一般的な負の reinforcement、時間経過だけの decay、未利用だけを理由にした減衰を導入しない。 + +## Contribution journal と exact reversal + +SQLite は soft-start の provisional contribution と各 confirmation contribution を独立した永続単位として保存する。一つの contribution は次を一体として持つ。 + +- 保存済み trace ID、source record ID、credited edge identity、初期 baseline weight +- credited edge へ実際に加算した正の actual delta +- その加算が同じ transaction で発生させた same-source sibling ごとの負の actual delta +- active / reversed 状態と reversal を行った outcome ID + +`corrected` と `rolled_back` は、outcome の node と保存済み credited path が一致する active contribution だけを対象にする。credited delta と、その同じ contribution に属する全 sibling normalization delta を一つの可逆単位として逆適用する。片側だけを戻さず、別 trace、別 contribution、別 source、uncredited edge、lexical path、zero-hop path は変更しない。 + +credited edge の逆適用は contribution の登録 baseline より下へ下げない。これは誤りに帰属された加算を取り消す境界であり、基礎 weight 以下の punitive update ではない。後から別 contribution が同じ edge へ加えた delta は保持する。candidate 有効時、active contribution を持つ edge は他 contribution の sibling normalization 対象から除外し、逆写像を曖昧にしない。 + +## Dormancy と reactivation + +`superseded` は帰属可能な credited relation edge を削除せず dormant として保存する。dormant edge は通常の graph activation に使う outgoing edge 集合から除外するが、edge、証拠、trace、contribution journal は保持する。 + +同じ保存済み credited path に後から `confirmed` が記録された場合、confirmation transaction の先頭で dormant 状態を解除する。receipt は `dormancy_changes` または `reactivated_edges` として old/new state を返す。再活性化だけで過去 contribution を再適用せず、新しい confirmation delta は既存 soft-start schedule に従う。 + +## Transaction、idempotency、receipt + +outcome row、node association、credited / sibling inverse、contribution state、dormancy state、idempotency receipt は一つの SQLite transaction に含める。途中で一つでも失敗した場合は outcome と graph mutation の両方を call 前へ戻す。同じ idempotency key と payload の replay は最初の receipt を返し、二重減算、二重 dormancy、二重 reactivation を行わない。restart 後も同じ journal と状態を使う。 + +core の `OutcomeReceipt` と optional MCP output は次を同じ意味と順序で公開する。 + +- `deactivation_applied` +- `reversed_contributions` と各 `mutations` +- `dormancy_changes` +- `reactivated_edges` + +既存 `reinforcement_applied` は正の confirmation update だけを表し、negative outcome では `False` のままとする。 + +## Result-free snapshot evaluation + +`outcome-driven-feedback-deactivation-v1` は fresh transaction-consistent private snapshot の clone 上で、`control` と `deactivation_candidate` を比較する。development / holdout は互いに異なる固定 case identity を使い、`corrected`、`rolled_back`、`superseded`、unattributed control を含む。protocol、fixture、schedule、gate、result schema、manifest、exclusive output path は registered result 不在の freeze commit で固定する。 + +result-free preflight は snapshot hash、integrity、schema、case capacity、registered baseline、relation path identity、same-source sibling capacity、privacy、writer / verifier round-trip、development / holdout output 不在を検証する。この Issue では development と holdout を実行しない。freeze が main へ squash mergeされた後の successor Issue が development を一度だけ実行し、全 hard gate 通過時だけ holdout を一度だけ実行する。観測後に query、case、ratio、schedule、metric、gate を変更しない。 + +public artifact は locator、capture時刻、logical source / snapshot hash、snapshot container hash、schema hash、row count、public node / edge identifier、query、baseline数値だけを持つ。private本文、credential、absolute private path、snapshot本体をcommitしない。source database、live config、private snapshotは実行を通して変更しない。 + +## 解釈境界 + +mechanics test と result-free preflight の通過は、candidate の有用性や local adoption を支持しない。successor development が不支持または判定不能なら holdout を開かず、現行 policy を維持する。holdoutまで全 gate を通過しても、支持範囲は固定 local snapshot と登録 schedule に限定し、library default や一般化は別判断とする。 + +## 関連 + +- [Requirements](requirements.md) +- [Confirmed-outcome feedback reinforcement](confirmed-outcome-feedback-reinforcement.md) +- [Baseline-aware soft-start snapshot evaluation](baseline-aware-soft-start-snapshot-evaluation.md) +- [Optional MCP Feedback Interface](optional-mcp-interface.md) +- [Decision Structure](Decision-Structure.md) +- [Issue #111](https://github.com/Liplus-Project/neuron-graph-rag/issues/111) diff --git a/docs/requirements.md b/docs/requirements.md index 6644e91..7c58c26 100644 --- a/docs/requirements.md +++ b/docs/requirements.md @@ -102,6 +102,8 @@ 83. soft-start feedback reinforcement は明示 opt-in の relation-only candidate とし、最初の新規 `used` で通常 bounded increment の固定 ratio 分だけを credited path へ適用する。最初の独立 `confirmed` は同じ schedule の残量を一回分の通常 increment まで補い、後続 confirmation は固定 decay ratio で加算する。used 時は sibling normalization を行わず、confirmation の actual delta だけを同一 source の uncredited sibling へ配分する。duplicate、lexical、zero-hop、別 source、uncredited edge、negative outcome は変更せず、candidate mechanics の合格だけで default や local serving policy を変更しない。 84. soft-start snapshot evaluation は、local source database を read-only URI と SQLite backup API で一度だけ transaction-consistent snapshot へ複製し、同じ snapshot のfresh clone上で `control`、`used_q3_s1`、`confirmed_r05_s1`、`soft_start_r025_r05_s1` の固定4 armを比較する。query、public node / edge identifier、outcome、3回のfresh trace schedule、checkpoint、metric、hard gate、exclusive output、snapshot / protocol hashを登録result生成前に固定する。source database、live config、snapshot本体、private本文、absolute private pathをpublic artifactへ含めず、developmentを一度だけ実行し、全hard gate通過時だけholdoutを一度開く。不支持または判定不能を保存してもprotocolを調整せず、local cutover、library default、external corpus、production qualityへ自動で一般化しない。 85. baseline-aware soft-start snapshot evaluation は、v1のprotocol、gate、observed result、private snapshotを変更、再実行、再集計、入力再利用せず、fresh snapshot、新規namespace、新規output、v1 observed developmentと異なるcredited edge identityを使う。各relation caseのinitial weight、reinforced count、evidence count、confirmation countを結果前に登録し、q3/s1のfirst mutationを`max(1, quorum - initial evidence count)`で導出する。導出不能、baseline不一致、event budget内のquorum capacity不足はregistered resultを作らずfailure reportで停止する。v2 protocolはregistered output不在のfreeze-only PRで固定し、そのsquash merge後の別Issueでdevelopmentを一度だけ実行する。全8 hard gate通過時だけholdoutを一度開き、支持結果もlocal cutover候補に限定してsource database、live config、library defaultを変更しない。 +86. outcome-driven feedback deactivation はsoft-startと同時にだけ有効化できるdefault-off candidateとする。provisional / confirmationごとにcredited加算と同時発生したsame-source sibling normalization減算を一つのsigned mutation journalへ永続化し、因果帰属できる`corrected` / `rolled_back`だけが未反転contributionを基礎weight未満へ下げずexact reversalする。`superseded`はedge、evidence、trace、outcomeを削除せずrelation edgeをdormantにして通常activationから除外し、同じ保存済みcredited pathの後続`confirmed`で再活性化する。duplicate、retry、restart、transaction failure、lexical、zero-hop、別source、uncredited edge、因果帰属不能outcomeは二重減算または局所外mutationを行わない。 +87. outcome-driven deactivation evaluation はcontrol / candidate、`corrected` / `rolled_back` / `superseded`、exact credited / sibling inverse、baseline floor、dormancy / reactivation、rank / locality、source isolation、exclusive outputを結果観測前に固定する。protocolはregistered output不在のfreeze-only PRで固定し、そのsquash merge後のsuccessor Issueでdevelopmentを一度だけ実行する。全hard gate通過時だけholdoutを一度開き、観測前後にquery、case、schedule、metric、gate、default、live configを変更しない。 75. v3 implementation、prompt、manifest、query override、schema、集約、path audit、hash規則、gate、stop rule、testsをresult-free commitでpushした後、development stage / 4 case packet / 12 responses / resultを各一度だけ生成する。 76. development全12 gate通過時だけholdout stageを一度生成し、異なるfresh 12 judgesで同じgateを評価する。packet、response、resultの上書き、観測後の規則変更、実LLM品質値のCI再生成を拒否する。 @@ -139,4 +141,5 @@ - [Historical source verification](historical-source-verification.md) が frozen manifest path の初回追加 commit、明示 full source commit ID、exact blob、ancestor、path、newline portability、fail-closed 境界を定義する。 - [Confirmed-outcome feedback reinforcement](confirmed-outcome-feedback-reinforcement.md) が confirmed-only と soft-start の明示 policy、永続 schedule、transaction、receipt、default-preserving boundary を定義する。 - [Soft-start snapshot evaluation](soft-start-snapshot-evaluation.md) が transaction-consistent private snapshot、固定4 arm、result-free freeze、privacy、one-time development、conditional holdout、local cutover境界を定義する。 +- [Outcome-driven feedback deactivation](outcome-driven-feedback-deactivation.md) がsigned contribution journal、exact reversal、dormancy / reactivation、default-off境界、result-free freezeを定義する。 - [Baseline-aware soft-start snapshot evaluation](baseline-aware-soft-start-snapshot-evaluation.md) がfresh baseline stateからのq3 boundary導出、v1 evidence isolation、capacity preflight、新規one-time result境界を定義する。 diff --git a/src/neuron_graph_rag/__init__.py b/src/neuron_graph_rag/__init__.py index a56e4f2..71fcd2d 100644 --- a/src/neuron_graph_rag/__init__.py +++ b/src/neuron_graph_rag/__init__.py @@ -3,14 +3,17 @@ from .models import ( ActivationPath, ConfirmedEdge, + ContributionMutation, CreditedPath, DocumentNode, + DormancyChange, FeedbackContractError, FeedbackEvidence, FeedbackReceipt, NormalizedSiblingEdge, OutcomeReceipt, PathStep, + ReversedContribution, SearchChannelHit, SearchChannelsResult, SearchChannelTrace, @@ -26,8 +29,10 @@ __all__ = [ "ActivationPath", "ConfirmedEdge", + "ContributionMutation", "CreditedPath", "DocumentNode", + "DormancyChange", "EngineConfig", "FeatureHashingEncoder", "FeedbackContractError", @@ -38,6 +43,7 @@ "NormalizedSiblingEdge", "OutcomeReceipt", "PathStep", + "ReversedContribution", "SearchChannelHit", "SearchChannelTrace", "SearchChannelsResult", diff --git a/src/neuron_graph_rag/config_provenance.py b/src/neuron_graph_rag/config_provenance.py index deb73db..b74ed78 100644 --- a/src/neuron_graph_rag/config_provenance.py +++ b/src/neuron_graph_rag/config_provenance.py @@ -17,6 +17,7 @@ "confirmation_decay_ratio", "soft_start_feedback_reinforcement", "soft_start_feedback_ratio", + "outcome_driven_feedback_deactivation", ) SEARCH_SURFACES = ("combined", "relation") @@ -42,6 +43,8 @@ def effective_config(config: EngineConfig) -> dict[str, dict[str, Any]]: "soft_start_feedback_reinforcement", "soft_start_feedback_ratio", } + if raw["outcome_driven_feedback_deactivation"] is False: + active_names.remove("outcome_driven_feedback_deactivation") return { "retrieval": { name: raw[name] diff --git a/src/neuron_graph_rag/evidence_feedback.py b/src/neuron_graph_rag/evidence_feedback.py index 293d6c5..4d8bd9b 100644 --- a/src/neuron_graph_rag/evidence_feedback.py +++ b/src/neuron_graph_rag/evidence_feedback.py @@ -26,6 +26,7 @@ class EngineConfig(BaseEngineConfig): confirmation_decay_ratio: float | None = None soft_start_feedback_reinforcement: bool = False soft_start_feedback_ratio: float | None = None + outcome_driven_feedback_deactivation: bool = False def __post_init__(self) -> None: BaseEngineConfig.__post_init__(self) @@ -41,6 +42,8 @@ def __post_init__(self) -> None: raise TypeError("confirmed_outcome_reinforcement must be a boolean") if not isinstance(self.soft_start_feedback_reinforcement, bool): raise TypeError("soft_start_feedback_reinforcement must be a boolean") + if not isinstance(self.outcome_driven_feedback_deactivation, bool): + raise TypeError("outcome_driven_feedback_deactivation must be a boolean") if ( self.confirmed_outcome_reinforcement and self.soft_start_feedback_reinforcement @@ -86,6 +89,14 @@ def __post_init__(self) -> None: raise ValueError( "soft_start_feedback_ratio requires soft_start_feedback_reinforcement" ) + if ( + self.outcome_driven_feedback_deactivation + and not self.soft_start_feedback_reinforcement + ): + raise ValueError( + "outcome_driven_feedback_deactivation requires soft-start feedback " + "reinforcement" + ) class NeuronGraphRAG(BaseNeuronGraphRAG): @@ -291,6 +302,16 @@ def confirmed_outcome_plan( trace_id, used_node_ids, require_candidate_marker=True ) + def deactivation_outcome_plan( + self, trace_id: str, used_node_ids: Iterable[str] + ) -> dict[str, Any]: + """Build the saved credited relation path used to attribute deactivation.""" + if not self.config.outcome_driven_feedback_deactivation: + raise ValueError("outcome-driven feedback deactivation is not enabled") + return self._relation_feedback_plan( + trace_id, used_node_ids, require_candidate_marker=True + ) + def _relation_feedback_plan( self, trace_id: str, @@ -381,6 +402,12 @@ def _relation_feedback_plan( for edge in self.store.outgoing_edges(source_id) if (edge.source_id, edge.target_id, edge.edge_type) not in credited_keys + and ( + not self.config.outcome_driven_feedback_deactivation + or not self.store.edge_has_active_contribution( + edge.source_id, edge.target_id, edge.edge_type + ) + ) ), self.config.sibling_feedback_normalization, ) diff --git a/src/neuron_graph_rag/feedback.py b/src/neuron_graph_rag/feedback.py index 9dac782..c0d2424 100644 --- a/src/neuron_graph_rag/feedback.py +++ b/src/neuron_graph_rag/feedback.py @@ -9,13 +9,16 @@ from .engine import NeuronGraphRAG from .models import ( ConfirmedEdge, + ContributionMutation, CreditedPath, + DormancyChange, FeedbackEvidence, FeedbackReceipt, NormalizedSiblingEdge, OutcomeReceipt, PathStep, ReinforcedEdge, + ReversedContribution, SourceUseEvent, SourceUseEventReceipt, SourceUseReceipt, @@ -169,6 +172,13 @@ def record_outcome( getattr(self.engine.config, "confirmed_outcome_reinforcement", False) or getattr(self.engine.config, "soft_start_feedback_reinforcement", False) ) + deactivation_enabled = bool( + getattr( + self.engine.config, + "outcome_driven_feedback_deactivation", + False, + ) + ) if candidate_enabled and outcome == "confirmed": plan = self.engine.confirmed_outcome_plan(trace_id, ordered_node_ids) record_confirmed = ( @@ -200,6 +210,26 @@ def record_outcome( credited_paths=plan["credited_paths"], **candidate_options, ) + elif deactivation_enabled and outcome in { + "corrected", + "rolled_back", + "superseded", + }: + plan = self.engine.deactivation_outcome_plan( + trace_id, ordered_node_ids + ) + stored = self.engine.store.record_deactivation_outcome( + idempotency_key=idempotency_key, + payload_json=payload_json, + outcome_id=outcome_id, + trace_id=trace_id, + node_ids=ordered_node_ids, + outcome=outcome, + summary=summary, + external_ref=external_ref, + recorded_at=recorded_at, + credited_paths=plan["credited_paths"], + ) else: stored = self.engine.store.record_outcome( idempotency_key=idempotency_key, @@ -258,6 +288,51 @@ def record_outcome( ) for edge in stored.get("normalized_sibling_edges", []) ), + bool(stored.get("deactivation_applied", False)), + tuple( + ReversedContribution( + str(item["contribution_id"]), + str(item["contribution_kind"]), + str(item["source_record_id"]), + str(item["source_id"]), + str(item["target_id"]), + str(item["edge_type"]), + float(item["credited_delta"]), + tuple( + ContributionMutation( + str(mutation["mutation_role"]), + str(mutation["source_id"]), + str(mutation["target_id"]), + str(mutation["edge_type"]), + float(mutation["actual_delta"]), + float(mutation["old_weight"]), + float(mutation["new_weight"]), + ) + for mutation in item["mutations"] + ), + ) + for item in stored.get("reversed_contributions", []) + ), + tuple( + DormancyChange( + str(item["source_id"]), + str(item["target_id"]), + str(item["edge_type"]), + bool(item["old_dormant"]), + bool(item["new_dormant"]), + ) + for item in stored.get("dormancy_changes", []) + ), + tuple( + DormancyChange( + str(item["source_id"]), + str(item["target_id"]), + str(item["edge_type"]), + bool(item["old_dormant"]), + bool(item["new_dormant"]), + ) + for item in stored.get("reactivated_edges", []) + ), ) @staticmethod diff --git a/src/neuron_graph_rag/models.py b/src/neuron_graph_rag/models.py index 5a019c2..ec3f9c8 100644 --- a/src/neuron_graph_rag/models.py +++ b/src/neuron_graph_rag/models.py @@ -267,6 +267,38 @@ class CreditedPath: steps: tuple[PathStep, ...] +@dataclass(frozen=True, slots=True) +class ContributionMutation: + mutation_role: str + source_id: str + target_id: str + edge_type: str + actual_delta: float + old_weight: float + new_weight: float + + +@dataclass(frozen=True, slots=True) +class ReversedContribution: + contribution_id: str + contribution_kind: str + source_record_id: str + source_id: str + target_id: str + edge_type: str + credited_delta: float + mutations: tuple[ContributionMutation, ...] + + +@dataclass(frozen=True, slots=True) +class DormancyChange: + source_id: str + target_id: str + edge_type: str + old_dormant: bool + new_dormant: bool + + @dataclass(frozen=True, slots=True) class OutcomeReceipt: outcome_id: str @@ -278,6 +310,10 @@ class OutcomeReceipt: confirmations: tuple[ConfirmedEdge, ...] = () credited_paths: tuple[CreditedPath, ...] = () normalized_sibling_edges: tuple[NormalizedSiblingEdge, ...] = () + deactivation_applied: bool = False + reversed_contributions: tuple[ReversedContribution, ...] = () + dormancy_changes: tuple[DormancyChange, ...] = () + reactivated_edges: tuple[DormancyChange, ...] = () class FeedbackContractError(ValueError): diff --git a/src/neuron_graph_rag/outcome_feedback_deactivation_evaluation.py b/src/neuron_graph_rag/outcome_feedback_deactivation_evaluation.py new file mode 100644 index 0000000..f4a7907 --- /dev/null +++ b/src/neuron_graph_rag/outcome_feedback_deactivation_evaluation.py @@ -0,0 +1,641 @@ +from __future__ import annotations + +import hashlib +import json +import os +import re +import shutil +import sqlite3 +from collections.abc import Mapping, Sequence +from dataclasses import asdict +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import Any + +from .corpus_integrity import verify_manifest_source_hashes +from .evidence_feedback import EngineConfig, NeuronGraphRAG +from .feedback import FeedbackLedger +from .models import SourceUseEvent + + +ROOT = Path(__file__).resolve().parents[2] +STEM = "outcome_feedback_deactivation_v1" +PROTOCOL_ID = "outcome-driven-feedback-deactivation-v1" +MANIFEST_PATH = ROOT / "tests" / "fixtures" / f"{STEM}.manifest.json" +STAGES = ("development", "holdout") +ARMS = ("control", "deactivation_candidate") +EDGE_FIELDS = ("source_id", "target_id", "edge_type") +_ABSOLUTE_PRIVATE_PATH = re.compile( + r"(?:[A-Za-z]:[\\/]|/Users/|/home/|\\\\[^\\]+\\[^\\]+)" +) +_CREDENTIAL = re.compile( + r"(?i)(?:gh[pousr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,}|" + r"(?:token|password|secret|api[_-]?key)\s*[:=]\s*[^\s,}]+)" +) + + +def _encoded(payload: Any) -> bytes: + return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8") + + +def _sha256_bytes(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest() + + +def _sha256(path: Path) -> str: + return _sha256_bytes(path.read_bytes()) + + +def read_json(path: Path) -> Any: + raw = path.read_bytes() + text = raw.decode("utf-8", errors="strict") + payload = json.loads(text) + if text != json.dumps(payload, ensure_ascii=False, indent=2) + "\n": + raise ValueError(f"non-canonical JSON artifact: {path}") + return payload + + +def write_json_exclusive(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(_encoded(payload)) + stream.flush() + os.fsync(stream.fileno()) + except BaseException: + path.unlink(missing_ok=True) + raise + + +def assert_public_payload(payload: Any) -> None: + def walk(value: Any, path: tuple[str, ...]) -> None: + if isinstance(value, Mapping): + for key, item in value.items(): + name = str(key) + if name.lower() in { + "node_text", + "document_text", + "raw_text", + "private_path", + "source_path", + "snapshot_path", + }: + raise ValueError(f"private field is forbidden: {'.'.join((*path, name))}") + walk(item, (*path, name)) + elif isinstance(value, (list, tuple)): + for index, item in enumerate(value): + walk(item, (*path, str(index))) + elif isinstance(value, str): + if value.startswith(("/", "\\")) or _ABSOLUTE_PRIVATE_PATH.search(value): + raise ValueError(f"absolute private path is forbidden: {'.'.join(path)}") + if _CREDENTIAL.search(value): + raise ValueError(f"credential-shaped value is forbidden: {'.'.join(path)}") + + walk(payload, ()) + + +def _logical_hash(connection: sqlite3.Connection) -> str: + return _sha256_bytes("\n".join(connection.iterdump()).encode("utf-8")) + + +def _schema_identity(connection: sqlite3.Connection) -> dict[str, Any]: + rows = [ + {"name": str(row[0]), "sql": str(row[1])} + for row in connection.execute( + "SELECT name, sql FROM sqlite_master " + "WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name" + ) + ] + return { + "schema_sha256": _sha256_bytes(_encoded(rows)), + "table_names": [row["name"] for row in rows], + } + + +def acquire_transactional_snapshot(source: Path, destination: Path) -> dict[str, Any]: + """Create one exclusive read-only SQLite backup and prove logical stability.""" + source = source.resolve(strict=True) + destination = destination.resolve(strict=False) + if source == destination: + raise ValueError("source and snapshot must differ") + destination.parent.mkdir(parents=True, exist_ok=True) + reservation = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + os.close(reservation) + captured_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat() + try: + source_connection = sqlite3.connect(source.as_uri() + "?mode=ro", uri=True) + snapshot_connection = sqlite3.connect(destination) + post_connection = sqlite3.connect(":memory:") + try: + source_connection.execute("PRAGMA query_only = ON") + source_connection.backup(snapshot_connection) + snapshot_connection.commit() + source_before = _logical_hash(snapshot_connection) + source_connection.backup(post_connection) + source_after = _logical_hash(post_connection) + if source_before != source_after: + raise RuntimeError("source database changed during snapshot acquisition") + if snapshot_connection.execute("PRAGMA integrity_check").fetchone()[0] != "ok": + raise RuntimeError("snapshot integrity check failed") + schema = _schema_identity(snapshot_connection) + counts = { + name: int( + snapshot_connection.execute(f"SELECT count(*) FROM {name}").fetchone()[0] + ) + for name in ( + "nodes", + "edges", + "retrievals", + "success_feedback", + "delayed_outcomes", + ) + } + finally: + post_connection.close() + snapshot_connection.close() + source_connection.close() + provenance = { + "source_locator": "local_codex_ngr_database", + "source_access": "sqlite-uri-mode-ro-query-only", + "capture_method": "sqlite-backup-api", + "captured_at": captured_at, + "source_logical_sha256_before": source_before, + "source_logical_sha256_after": source_after, + "snapshot_logical_sha256": source_before, + "snapshot_sha256": _sha256(destination), + "snapshot_size": destination.stat().st_size, + **schema, + "row_counts": counts, + } + assert_public_payload(provenance) + return provenance + except BaseException: + destination.unlink(missing_ok=True) + raise + + +def prove_writer_verifier_round_trip(path: Path) -> None: + payload = { + "protocol_id": PROTOCOL_ID, + "stage": "placeholder", + "gate_order": ["zulu-placeholder", "alpha-placeholder", "mike-placeholder"], + } + write_json_exclusive(path, payload) + try: + if read_json(path) != payload: + raise ValueError("placeholder semantics changed") + finally: + path.unlink(missing_ok=True) + + +def _load_current_protocol() -> tuple[dict[str, Any], dict[str, Any]]: + manifest = read_json(MANIFEST_PATH) + for relative, expected in manifest["artifact_sha256"].items(): + if _sha256(ROOT / relative) != expected: + raise ValueError(f"current artifact hash mismatch: {relative}") + artifacts = { + name: read_json(ROOT / relative) + for name, relative in manifest["protocol_artifacts"].items() + } + assert_public_payload(manifest) + assert_public_payload(artifacts) + return manifest, artifacts + + +def _load_registered_protocol() -> tuple[dict[str, Any], dict[str, Any]]: + manifest = read_json(MANIFEST_PATH) + registered = verify_manifest_source_hashes( + ROOT, MANIFEST_PATH, manifest["artifact_sha256"] + ) + artifacts = { + name: json.loads(registered.artifact_bytes[relative].decode("utf-8", errors="strict")) + for name, relative in manifest["protocol_artifacts"].items() + } + assert_public_payload(manifest) + assert_public_payload(artifacts) + return manifest, artifacts + + +def _edge_key(value: Mapping[str, Any] | Sequence[str]) -> str: + if isinstance(value, Mapping): + return "|".join(str(value[field]) for field in EDGE_FIELDS) + return "|".join(str(item) for item in value) + + +def _config(candidate: bool) -> EngineConfig: + return EngineConfig( + soft_start_feedback_reinforcement=True, + soft_start_feedback_ratio=0.25, + confirmation_decay_ratio=0.5, + sibling_feedback_normalization=1.0, + outcome_driven_feedback_deactivation=candidate, + ) + + +def _edge_state(engine: NeuronGraphRAG) -> dict[str, dict[str, Any]]: + return { + _edge_key((edge.source_id, edge.target_id, edge.edge_type)): { + "weight": edge.weight, + "reinforced_count": edge.reinforced_count, + "dormant": engine.store.edge_is_dormant( + edge.source_id, edge.target_id, edge.edge_type + ), + } + for edge in engine.store.list_edges() + } + + +def _select_trace(engine: NeuronGraphRAG, case: Mapping[str, Any], now: float) -> str: + channels = engine.search_channels(str(case["query"]), limit=21, now=now) + surface = str(case["search_surface"]) + trace = channels.relation if surface == "relation" else channels.lexical + target = str(case["used_node_id"]) + if target not in {hit.node.node_id for hit in trace.hits}: + raise RuntimeError("registered node is absent from its search trace") + if surface == "relation": + selected = max( + engine.store.retrieval_paths(trace.trace_id, target), + key=lambda path: (float(path["contribution"]), str(path["seed_id"])), + default=None, + ) + expected = tuple(str(case["credited_edge"][field]) for field in EDGE_FIELDS) + actual = () if selected is None else tuple( + (str(step["source_id"]), str(step["target_id"]), str(step["edge_type"])) + for step in selected["steps"] + ) + if actual != (expected,): + raise RuntimeError("registered credited edge is not the selected path") + return trace.trace_id + + +def _used(ledger: FeedbackLedger, trace_id: str, node_id: str, key: str, now: float) -> Any: + return ledger.record_source_use( + trace_id, + tuple(SourceUseEvent(node_id, stage) for stage in ("selected", "validated", "used")), + idempotency_key=key, + now=now, + ) + + +def _receipt(value: Any) -> dict[str, Any]: + return asdict(value) + + +def _observe_case( + snapshot: Path, + arm_id: str, + case: Mapping[str, Any], + case_index: int, +) -> dict[str, Any]: + candidate = arm_id == "deactivation_candidate" + with TemporaryDirectory() as directory: + clone = Path(directory) / "case.sqlite" + shutil.copyfile(snapshot, clone) + clock = 100_000.0 + case_index * 1_000.0 + reopened: NeuronGraphRAG | None = None + with NeuronGraphRAG(clone, config=_config(candidate)) as engine: + baseline = _edge_state(engine) + ledger = FeedbackLedger(engine) + trace_id = _select_trace(engine, case, clock) + used = _used( + ledger, + trace_id, + str(case["used_node_id"]), + f"{arm_id}-{case['case_id']}-used", + clock + 1.0, + ) + after_used = _edge_state(engine) + confirmed: dict[str, Any] | None = None + atomicity = True + if case["case_role"] in {"corrected", "rolled_back"}: + confirmed_receipt = ledger.record_outcome( + trace_id, + [str(case["used_node_id"])], + "confirmed", + "registered positive outcome", + idempotency_key=f"{arm_id}-{case['case_id']}-confirmed", + now=clock + 2.0, + ) + confirmed = _receipt(confirmed_receipt) + if case["case_role"] == "rolled_back": + engine.close() + reopened = NeuronGraphRAG(clone, config=_config(candidate)) + engine = reopened + ledger = FeedbackLedger(engine) + before_negative = _edge_state(engine) + if candidate: + engine.store.connection.execute( + "CREATE TEMP TRIGGER fail_deactivation BEFORE UPDATE OF active " + "ON feedback_contributions WHEN NEW.active = 0 " + "BEGIN SELECT RAISE(ABORT, 'atomicity probe'); END" + ) + try: + ledger.record_outcome( + trace_id, + [str(case["used_node_id"])], + str(case["negative_outcome"]), + "atomicity probe", + idempotency_key=f"{arm_id}-{case['case_id']}-atomicity", + now=clock + 2.5, + ) + except sqlite3.IntegrityError: + pass + else: + atomicity = False + finally: + engine.store.connection.execute("DROP TRIGGER fail_deactivation") + atomicity = atomicity and _edge_state(engine) == before_negative + else: + before_negative = _edge_state(engine) + + negative = ledger.record_outcome( + trace_id, + [str(case["used_node_id"])], + str(case["negative_outcome"]), + "registered negative outcome", + idempotency_key=f"{arm_id}-{case['case_id']}-negative", + now=clock + 3.0, + ) + replay = ledger.record_outcome( + trace_id, + [str(case["used_node_id"])], + str(case["negative_outcome"]), + "registered negative outcome", + idempotency_key=f"{arm_id}-{case['case_id']}-negative", + now=clock + 9.0, + ) + after_negative = _edge_state(engine) + dormant_hidden = None + reactivated: dict[str, Any] | None = None + if case["case_role"] == "superseded": + edge = case["credited_edge"] + dormant_hidden = _edge_key(edge) not in { + _edge_key((item.source_id, item.target_id, item.edge_type)) + for item in engine.store.outgoing_edges(str(edge["source_id"])) + } + reactivated_receipt = ledger.record_outcome( + trace_id, + [str(case["used_node_id"])], + "confirmed", + "registered reactivation outcome", + idempotency_key=f"{arm_id}-{case['case_id']}-reactivated", + now=clock + 4.0, + ) + reactivated = _receipt(reactivated_receipt) + final = _edge_state(engine) + if reopened is not None: + reopened.close() + return { + "case_id": case["case_id"], + "case_role": case["case_role"], + "baseline": baseline, + "after_used": after_used, + "before_negative": before_negative, + "after_negative": after_negative, + "final": final, + "used": _receipt(used), + "confirmed": confirmed, + "negative": _receipt(negative), + "idempotency_replay_equal": replay == negative, + "atomicity_probe_passed": atomicity, + "dormant_hidden": dormant_hidden, + "reactivated": reactivated, + } + + +def evaluate_gates( + preflight: Mapping[str, Any], + cases: Sequence[Mapping[str, Any]], + arms: Mapping[str, Sequence[Mapping[str, Any]]], +) -> dict[str, bool]: + registered = {str(case["case_id"]): case for case in cases} + candidate = {str(item["case_id"]): item for item in arms["deactivation_candidate"]} + control = {str(item["case_id"]): item for item in arms["control"]} + + def exact(role: str) -> bool: + case = next(item for item in cases if item["case_role"] == role) + key = _edge_key(case["credited_edge"]) + observed = candidate[str(case["case_id"])] + receipt = observed["negative"] + mutation_roles = { + mutation["mutation_role"] + for contribution in receipt["reversed_contributions"] + for mutation in contribution["mutations"] + } + return ( + receipt["deactivation_applied"] + and mutation_roles == {"credited", "sibling"} + and observed["after_negative"][key] == observed["baseline"][key] + and observed["after_negative"] == observed["baseline"] + and control[str(case["case_id"])]["after_negative"] + == control[str(case["case_id"])]["before_negative"] + ) + + superseded_case = next(item for item in cases if item["case_role"] == "superseded") + superseded = candidate[str(superseded_case["case_id"])] + edge_key = _edge_key(superseded_case["credited_edge"]) + dormancy = ( + superseded["negative"]["deactivation_applied"] + and superseded["after_negative"][edge_key]["dormant"] + and superseded["dormant_hidden"] is True + and superseded["reactivated"] is not None + and len(superseded["reactivated"]["reactivated_edges"]) == 1 + and not superseded["final"][edge_key]["dormant"] + ) + all_cases = [item for values in arms.values() for item in values] + unattributed = [item for item in all_cases if item["case_role"] == "unattributed"] + locality = all( + item["after_negative"] == item["before_negative"] + for item in unattributed + ) and all( + item["idempotency_replay_equal"] for item in all_cases + ) + baseline_floor = all( + all( + state["weight"] >= item["baseline"][key]["weight"] + for key, state in item["after_negative"].items() + if key in item["baseline"] + ) + for item in candidate.values() + ) + control_audit_only = all( + not item["negative"]["deactivation_applied"] + for item in control.values() + ) + return { + "protocol-and-preflight-integrity": bool(preflight["passed"]), + "corrected-exact-contribution-reversal": exact("corrected"), + "rolled-back-exact-contribution-reversal": exact("rolled_back"), + "superseded-dormancy-and-reactivation": dormancy, + "baseline-floor-without-punitive-update": baseline_floor, + "idempotency-restart-and-transaction-atomicity": all( + item["atomicity_probe_passed"] and item["idempotency_replay_equal"] + for item in all_cases + ), + "credited-sibling-locality-and-controls": locality and control_audit_only, + "source-snapshot-privacy-and-exclusive-output": bool( + preflight["snapshot_unchanged"] + ), + } + + +def _preflight( + snapshot: Path, + manifest: Mapping[str, Any], + artifacts: Mapping[str, Any], + stage: str, +) -> dict[str, Any]: + snapshot_before = _sha256(snapshot) + cases = artifacts["fixture"]["stages"][stage] + with sqlite3.connect(snapshot.as_uri() + "?mode=ro", uri=True) as connection: + connection.execute("PRAGMA query_only = ON") + schema = _schema_identity(connection) + logical = _logical_hash(connection) + integrity = connection.execute("PRAGMA integrity_check").fetchone()[0] + node_ids = {str(row[0]) for row in connection.execute("SELECT node_id FROM nodes")} + edge_rows = { + _edge_key(row[:3]): (float(row[3]), int(row[4])) + for row in connection.execute( + "SELECT source_id, target_id, edge_type, weight, reinforced_count FROM edges" + ) + } + baseline_match = all( + edge_rows.get(_edge_key(case["credited_edge"])) + == ( + float(case["registered_initial_state"]["weight"]), + int(case["registered_initial_state"]["reinforced_count"]), + ) + for case in cases + if case["search_surface"] == "relation" + ) + sibling_capacity = all( + int( + connection.execute( + "SELECT count(*) FROM edges WHERE source_id = ? AND NOT " + "(target_id = ? AND edge_type = ?)", + ( + case["credited_edge"]["source_id"], + case["credited_edge"]["target_id"], + case["credited_edge"]["edge_type"], + ), + ).fetchone()[0] + ) + > 0 + for case in cases + if case["case_role"] in {"corrected", "rolled_back"} + ) + trace_eligibility = True + try: + with TemporaryDirectory() as directory: + clone = Path(directory) / "preflight.sqlite" + shutil.copyfile(snapshot, clone) + with NeuronGraphRAG(clone, config=_config(True)) as engine: + for index, case in enumerate(cases): + _select_trace(engine, case, 80_000.0 + index) + except Exception: + trace_eligibility = False + registered_nodes = { + str(case["used_node_id"]) + for case in cases + } | { + str(case["credited_edge"][field]) + for case in cases + if case["search_surface"] == "relation" + for field in ("source_id", "target_id") + } + output_absent = not (ROOT / manifest["outputs"][stage]).exists() + holdout_absent = not (ROOT / manifest["outputs"]["holdout"]).exists() + checks = { + "snapshot_container_hash": snapshot_before == manifest["snapshot"]["snapshot_sha256"], + "snapshot_logical_hash": logical == manifest["snapshot"]["snapshot_logical_sha256"], + "snapshot_integrity": integrity == "ok", + "snapshot_schema": schema == { + "schema_sha256": manifest["snapshot"]["schema_sha256"], + "table_names": manifest["snapshot"]["table_names"], + }, + "registered_nodes": registered_nodes <= node_ids, + "registered_baseline": baseline_match, + "same_source_sibling_capacity": sibling_capacity, + "trace_and_path_identity": trace_eligibility, + "arm_order": [arm["arm_id"] for arm in artifacts["schedule"]["arms"]] == list(ARMS), + "case_identity": len({case["case_id"] for case in cases}) == len(cases), + "registered_output_absent": output_absent, + "holdout_absent_before_development": stage != "development" or holdout_absent, + "privacy": True, + "placeholder_round_trip": artifacts["audit"]["placeholder_round_trip_passed"] is True, + } + return { + "checks": checks, + "passed": all(checks.values()), + "snapshot_sha256_before": snapshot_before, + "snapshot_sha256_after": _sha256(snapshot), + "snapshot_unchanged": _sha256(snapshot) == snapshot_before, + } + + +def preflight_snapshot(snapshot: Path) -> dict[str, Any]: + manifest, artifacts = _load_current_protocol() + reports = { + stage: _preflight(snapshot, manifest, artifacts, stage) for stage in STAGES + } + return { + "protocol_id": PROTOCOL_ID, + "stages": reports, + "passed": all(report["passed"] for report in reports.values()), + } + + +def run_registered_stage(stage: str, snapshot: Path) -> Path: + if stage not in STAGES: + raise ValueError("unknown stage") + manifest, artifacts = _load_registered_protocol() + preflight = _preflight(snapshot, manifest, artifacts, stage) + if not preflight["passed"]: + raise RuntimeError("protocol preflight failed; no registered result was written") + if stage == "holdout": + development = ROOT / manifest["outputs"]["development"] + if not development.exists() or not verify_registered_result("development"): + raise RuntimeError("development must pass before holdout") + cases = artifacts["fixture"]["stages"][stage] + observed = { + arm: [ + _observe_case(snapshot, arm, case, index) + for index, case in enumerate(cases) + ] + for arm in ARMS + } + gates = evaluate_gates(preflight, cases, observed) + gate_order = [item["gate_id"] for item in artifacts["gate"]["gates"]] + if list(gates) != gate_order: + raise RuntimeError("gate order differs from the frozen contract") + payload = { + "protocol_id": PROTOCOL_ID, + "stage": stage, + "snapshot_sha256": _sha256(snapshot), + "arms": observed, + "hard_gates": gates, + "all_hard_gates_pass": all(gates.values()), + } + assert_public_payload(payload) + output = ROOT / manifest["outputs"][stage] + write_json_exclusive(output, payload) + verify_registered_result(stage) + return output + + +def verify_registered_result(stage: str) -> bool: + if stage not in STAGES: + raise ValueError("unknown stage") + manifest, artifacts = _load_registered_protocol() + payload = read_json(ROOT / manifest["outputs"][stage]) + assert_public_payload(payload) + gate_order = [item["gate_id"] for item in artifacts["gate"]["gates"]] + if payload["protocol_id"] != PROTOCOL_ID or payload["stage"] != stage: + raise ValueError("registered result identity mismatch") + if list(payload["hard_gates"]) != gate_order: + raise ValueError("registered gate order mismatch") + if payload["all_hard_gates_pass"] != all(payload["hard_gates"].values()): + raise ValueError("registered aggregate gate mismatch") + return bool(payload["all_hard_gates_pass"]) diff --git a/src/neuron_graph_rag/storage.py b/src/neuron_graph_rag/storage.py index 1d5ac09..5b4b076 100644 --- a/src/neuron_graph_rag/storage.py +++ b/src/neuron_graph_rag/storage.py @@ -218,6 +218,54 @@ def _create_schema(self) -> None: FOREIGN KEY (source_id, target_id, edge_type) REFERENCES edges(source_id, target_id, edge_type) ON DELETE CASCADE ); + + CREATE TABLE IF NOT EXISTS feedback_contributions ( + contribution_id TEXT PRIMARY KEY, + contribution_kind TEXT NOT NULL + CHECK(contribution_kind IN ('soft_start_provisional', 'soft_start_confirmation')), + source_record_id TEXT NOT NULL, + trace_id TEXT NOT NULL REFERENCES retrievals(trace_id) ON DELETE CASCADE, + source_id TEXT NOT NULL, + target_id TEXT NOT NULL, + edge_type TEXT NOT NULL, + baseline_weight REAL NOT NULL CHECK(baseline_weight >= 0.0), + credited_delta REAL NOT NULL CHECK(credited_delta > 0.0), + reinforced_count_delta INTEGER NOT NULL CHECK(reinforced_count_delta >= 0), + active INTEGER NOT NULL DEFAULT 1 CHECK(active IN (0, 1)), + reversed_by_outcome_id TEXT REFERENCES delayed_outcomes(outcome_id), + created_at REAL NOT NULL, + UNIQUE(contribution_kind, source_record_id, source_id, target_id, edge_type), + FOREIGN KEY (source_id, target_id, edge_type) + REFERENCES edges(source_id, target_id, edge_type) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS feedback_contribution_mutations ( + contribution_id TEXT NOT NULL + REFERENCES feedback_contributions(contribution_id) ON DELETE CASCADE, + mutation_role TEXT NOT NULL CHECK(mutation_role IN ('credited', 'sibling')), + source_id TEXT NOT NULL, + target_id TEXT NOT NULL, + edge_type TEXT NOT NULL, + actual_delta REAL NOT NULL CHECK(actual_delta != 0.0), + PRIMARY KEY ( + contribution_id, mutation_role, source_id, target_id, edge_type + ), + FOREIGN KEY (source_id, target_id, edge_type) + REFERENCES edges(source_id, target_id, edge_type) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS relation_edge_dormancy ( + source_id TEXT NOT NULL, + target_id TEXT NOT NULL, + edge_type TEXT NOT NULL, + dormant INTEGER NOT NULL CHECK(dormant IN (0, 1)), + outcome_id TEXT NOT NULL REFERENCES delayed_outcomes(outcome_id), + trace_id TEXT NOT NULL REFERENCES retrievals(trace_id) ON DELETE CASCADE, + updated_at REAL NOT NULL, + PRIMARY KEY (source_id, target_id, edge_type), + FOREIGN KEY (source_id, target_id, edge_type) + REFERENCES edges(source_id, target_id, edge_type) ON DELETE CASCADE + ); """ ) self.connection.commit() @@ -283,14 +331,126 @@ def list_edges(self) -> list[TypedEdge]: def outgoing_edges(self, node_id: str) -> list[TypedEdge]: rows = self.connection.execute( """ - SELECT * FROM edges - WHERE source_id = ? - ORDER BY target_id, edge_type + SELECT edges.* FROM edges + LEFT JOIN relation_edge_dormancy AS dormancy + ON dormancy.source_id = edges.source_id + AND dormancy.target_id = edges.target_id + AND dormancy.edge_type = edges.edge_type + WHERE edges.source_id = ? AND COALESCE(dormancy.dormant, 0) = 0 + ORDER BY edges.target_id, edges.edge_type """, (node_id,), ).fetchall() return [self._edge_from_row(row) for row in rows] + def edge_has_active_contribution( + self, source_id: str, target_id: str, edge_type: str + ) -> bool: + return self.connection.execute( + """ + SELECT 1 FROM feedback_contributions + WHERE source_id = ? AND target_id = ? AND edge_type = ? AND active = 1 + LIMIT 1 + """, + (source_id, target_id, edge_type), + ).fetchone() is not None + + def edge_is_dormant( + self, source_id: str, target_id: str, edge_type: str + ) -> bool: + row = self.connection.execute( + """ + SELECT dormant FROM relation_edge_dormancy + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + (source_id, target_id, edge_type), + ).fetchone() + return row is not None and bool(row["dormant"]) + + @staticmethod + def _contribution_id( + contribution_kind: str, + source_record_id: str, + source_id: str, + target_id: str, + edge_type: str, + ) -> str: + value = "\0".join( + (contribution_kind, source_record_id, source_id, target_id, edge_type) + ) + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + @classmethod + def _insert_contribution( + cls, + connection: sqlite3.Connection, + *, + contribution_kind: str, + source_record_id: str, + trace_id: str, + source_id: str, + target_id: str, + edge_type: str, + baseline_weight: float, + credited_delta: float, + created_at: float, + sibling_mutations: Iterable[tuple[str, str, str, float]] = (), + ) -> str | None: + if credited_delta <= 0.0: + return None + contribution_id = cls._contribution_id( + contribution_kind, source_record_id, source_id, target_id, edge_type + ) + connection.execute( + """ + INSERT INTO feedback_contributions( + contribution_id, contribution_kind, source_record_id, trace_id, + source_id, target_id, edge_type, baseline_weight, credited_delta, + reinforced_count_delta, active, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, 1, ?) + """, + ( + contribution_id, + contribution_kind, + source_record_id, + trace_id, + source_id, + target_id, + edge_type, + baseline_weight, + credited_delta, + created_at, + ), + ) + connection.execute( + """ + INSERT INTO feedback_contribution_mutations( + contribution_id, mutation_role, source_id, target_id, edge_type, + actual_delta + ) VALUES (?, 'credited', ?, ?, ?, ?) + """, + (contribution_id, source_id, target_id, edge_type, credited_delta), + ) + for sibling_source, sibling_target, sibling_type, reduction in sibling_mutations: + if reduction <= 0.0: + continue + connection.execute( + """ + INSERT INTO feedback_contribution_mutations( + contribution_id, mutation_role, source_id, target_id, edge_type, + actual_delta + ) VALUES (?, 'sibling', ?, ?, ?, ?) + """, + ( + contribution_id, + sibling_source, + sibling_target, + sibling_type, + -reduction, + ), + ) + return contribution_id + def edge(self, source_id: str, target_id: str, edge_type: str) -> TypedEdge: row = self.connection.execute( """ @@ -884,6 +1044,20 @@ def apply_soft_start_feedback( created_at, ), ) + self._insert_contribution( + connection, + contribution_kind="soft_start_provisional", + source_record_id=feedback_id, + trace_id=trace_id, + source_id=source_id, + target_id=target_id, + edge_type=edge_type, + baseline_weight=( + old_weight if state is None else float(state["initial_weight"]) + ), + credited_delta=actual_delta, + created_at=created_at, + ) return reinforced def record_source_use( @@ -1267,6 +1441,11 @@ def record_soft_start_confirmed_outcome( payload_hash = hashlib.sha256(payload_json.encode("utf-8")).hexdigest() confirmations: list[dict[str, Any]] = [] normalized: list[dict[str, Any]] = [] + contribution_specs: list[dict[str, Any]] = [] + sibling_reductions_by_source: dict[ + str, list[tuple[str, str, str, float]] + ] = {} + reactivated: list[dict[str, Any]] = [] with self.transaction() as connection: replay = self._idempotent_replay( connection, idempotency_key, "record_outcome", payload_hash @@ -1300,6 +1479,38 @@ def record_soft_start_confirmed_outcome( ).fetchone() if duplicate is not None: continue + dormant = connection.execute( + """ + SELECT dormant FROM relation_edge_dormancy + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + (source_id, target_id, edge_type), + ).fetchone() + if dormant is not None and bool(dormant["dormant"]): + connection.execute( + """ + UPDATE relation_edge_dormancy + SET dormant = 0, outcome_id = ?, trace_id = ?, updated_at = ? + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + ( + outcome_id, + trace_id, + recorded_at, + source_id, + target_id, + edge_type, + ), + ) + reactivated.append( + { + "source_id": source_id, + "target_id": target_id, + "edge_type": edge_type, + "old_dormant": True, + "new_dormant": False, + } + ) edge = connection.execute( """ SELECT weight FROM edges @@ -1411,6 +1622,15 @@ def record_soft_start_confirmed_outcome( "new_weight": new_weight, } ) + contribution_specs.append( + { + "source_id": source_id, + "target_id": target_id, + "edge_type": edge_type, + "baseline_weight": float(state["initial_weight"]), + "credited_delta": actual_delta, + } + ) reinforced_increase_by_source[source_id] = ( reinforced_increase_by_source.get(source_id, 0.0) + actual_delta ) @@ -1451,6 +1671,40 @@ def record_soft_start_confirmed_outcome( "new_weight": new_weight, } ) + sibling_reductions_by_source.setdefault(source_id, []).append( + ( + sibling_source, + target_id, + edge_type, + old_weight - new_weight, + ) + ) + for spec in contribution_specs: + source_id = str(spec["source_id"]) + total = reinforced_increase_by_source.get(source_id, 0.0) + share = float(spec["credited_delta"]) / total if total > 0.0 else 0.0 + self._insert_contribution( + connection, + contribution_kind="soft_start_confirmation", + source_record_id=outcome_id, + trace_id=trace_id, + source_id=source_id, + target_id=str(spec["target_id"]), + edge_type=str(spec["edge_type"]), + baseline_weight=float(spec["baseline_weight"]), + credited_delta=float(spec["credited_delta"]), + created_at=recorded_at, + sibling_mutations=tuple( + ( + sibling_source, + sibling_target, + sibling_type, + reduction * share, + ) + for sibling_source, sibling_target, sibling_type, reduction + in sibling_reductions_by_source.get(source_id, []) + ), + ) result = { "outcome_id": outcome_id, "trace_id": trace_id, @@ -1461,6 +1715,7 @@ def record_soft_start_confirmed_outcome( "confirmations": confirmations, "credited_paths": list(credited_paths), "normalized_sibling_edges": normalized, + "reactivated_edges": reactivated, } self._save_idempotent_result( connection, idempotency_key, "record_outcome", payload_hash, result @@ -1515,6 +1770,232 @@ def record_outcome( ) return result + def record_deactivation_outcome( + self, + *, + idempotency_key: str, + payload_json: str, + outcome_id: str, + trace_id: str, + node_ids: tuple[str, ...], + outcome: str, + summary: str, + external_ref: str | None, + recorded_at: float, + credited_paths: tuple[dict[str, object], ...], + ) -> dict[str, Any]: + """Atomically reverse attributable contributions or change edge dormancy.""" + if outcome not in {"corrected", "rolled_back", "superseded"}: + raise ValueError("deactivation outcome is not supported") + payload_hash = hashlib.sha256(payload_json.encode("utf-8")).hexdigest() + with self.transaction() as connection: + replay = self._idempotent_replay( + connection, idempotency_key, "record_outcome", payload_hash + ) + if replay is not None: + return replay + self._require_trace(connection, trace_id) + self._require_used_nodes(connection, trace_id, node_ids) + connection.execute( + """ + INSERT INTO delayed_outcomes( + outcome_id, trace_id, outcome, summary, external_ref, recorded_at + ) VALUES (?, ?, ?, ?, ?, ?) + """, + (outcome_id, trace_id, outcome, summary, external_ref, recorded_at), + ) + for node_id in node_ids: + connection.execute( + "INSERT INTO delayed_outcome_nodes(outcome_id, node_id) VALUES (?, ?)", + (outcome_id, node_id), + ) + + credited_keys = { + ( + str(step["source_id"]), + str(step["target_id"]), + str(step["edge_type"]), + ) + for path in credited_paths + for step in path["steps"] + } + reversed_contributions: list[dict[str, Any]] = [] + dormancy_changes: list[dict[str, Any]] = [] + if outcome in {"corrected", "rolled_back"}: + for source_id, target_id, edge_type in sorted(credited_keys): + contributions = connection.execute( + """ + SELECT * FROM feedback_contributions + WHERE trace_id = ? AND source_id = ? AND target_id = ? + AND edge_type = ? AND active = 1 + ORDER BY created_at DESC, contribution_id DESC + """, + (trace_id, source_id, target_id, edge_type), + ).fetchall() + for contribution in contributions: + mutations: list[dict[str, Any]] = [] + mutation_rows = connection.execute( + """ + SELECT * FROM feedback_contribution_mutations + WHERE contribution_id = ? + ORDER BY mutation_role, source_id, target_id, edge_type + """, + (contribution["contribution_id"],), + ).fetchall() + for mutation in mutation_rows: + edge = connection.execute( + """ + SELECT weight, reinforced_count FROM edges + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + ( + mutation["source_id"], + mutation["target_id"], + mutation["edge_type"], + ), + ).fetchone() + if edge is None: + raise KeyError("journaled contribution edge is absent") + old_weight = float(edge["weight"]) + signed_delta = float(mutation["actual_delta"]) + if str(mutation["mutation_role"]) == "credited": + new_weight = max( + float(contribution["baseline_weight"]), + old_weight - signed_delta, + ) + new_count = max( + 0, + int(edge["reinforced_count"]) + - int(contribution["reinforced_count_delta"]), + ) + connection.execute( + """ + UPDATE edges SET weight = ?, reinforced_count = ? + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + ( + new_weight, + new_count, + mutation["source_id"], + mutation["target_id"], + mutation["edge_type"], + ), + ) + else: + new_weight = old_weight - signed_delta + connection.execute( + """ + UPDATE edges SET weight = ? + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + ( + new_weight, + mutation["source_id"], + mutation["target_id"], + mutation["edge_type"], + ), + ) + mutations.append( + { + "mutation_role": str(mutation["mutation_role"]), + "source_id": str(mutation["source_id"]), + "target_id": str(mutation["target_id"]), + "edge_type": str(mutation["edge_type"]), + "actual_delta": signed_delta, + "old_weight": old_weight, + "new_weight": new_weight, + } + ) + connection.execute( + """ + UPDATE feedback_contributions + SET active = 0, reversed_by_outcome_id = ? + WHERE contribution_id = ? + """, + (outcome_id, contribution["contribution_id"]), + ) + reversed_contributions.append( + { + "contribution_id": str(contribution["contribution_id"]), + "contribution_kind": str(contribution["contribution_kind"]), + "source_record_id": str(contribution["source_record_id"]), + "source_id": str(contribution["source_id"]), + "target_id": str(contribution["target_id"]), + "edge_type": str(contribution["edge_type"]), + "credited_delta": float(contribution["credited_delta"]), + "mutations": mutations, + } + ) + else: + for source_id, target_id, edge_type in sorted(credited_keys): + attributable = connection.execute( + """ + SELECT 1 FROM feedback_contributions + WHERE trace_id = ? AND source_id = ? AND target_id = ? + AND edge_type = ? + LIMIT 1 + """, + (trace_id, source_id, target_id, edge_type), + ).fetchone() + if attributable is None: + continue + current = connection.execute( + """ + SELECT dormant FROM relation_edge_dormancy + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + (source_id, target_id, edge_type), + ).fetchone() + old_dormant = current is not None and bool(current["dormant"]) + connection.execute( + """ + INSERT INTO relation_edge_dormancy( + source_id, target_id, edge_type, dormant, + outcome_id, trace_id, updated_at + ) VALUES (?, ?, ?, 1, ?, ?, ?) + ON CONFLICT(source_id, target_id, edge_type) DO UPDATE SET + dormant = 1, + outcome_id = excluded.outcome_id, + trace_id = excluded.trace_id, + updated_at = excluded.updated_at + """, + ( + source_id, target_id, edge_type, + outcome_id, trace_id, recorded_at, + ), + ) + if not old_dormant: + dormancy_changes.append( + { + "source_id": source_id, + "target_id": target_id, + "edge_type": edge_type, + "old_dormant": False, + "new_dormant": True, + } + ) + result = { + "outcome_id": outcome_id, + "trace_id": trace_id, + "node_ids": list(node_ids), + "outcome": outcome, + "recorded_at": recorded_at, + "reinforcement_applied": False, + "deactivation_applied": bool( + reversed_contributions or dormancy_changes + ), + "confirmations": [], + "credited_paths": list(credited_paths), + "normalized_sibling_edges": [], + "reversed_contributions": reversed_contributions, + "dormancy_changes": dormancy_changes, + "reactivated_edges": [], + } + self._save_idempotent_result( + connection, idempotency_key, "record_outcome", payload_hash, result + ) + return result + @staticmethod def _require_used_nodes( connection: sqlite3.Connection, trace_id: str, node_ids: tuple[str, ...] diff --git a/src/neuron_graph_rag_mcp/server.py b/src/neuron_graph_rag_mcp/server.py index fcadb71..de40e53 100644 --- a/src/neuron_graph_rag_mcp/server.py +++ b/src/neuron_graph_rag_mcp/server.py @@ -88,6 +88,15 @@ "confirmed outcome's actual delta. Duplicate traces and idempotency retries do not " "reinforce twice; negative outcomes remain audit-only." ) +DEACTIVATION_OUTCOME_DESCRIPTION = ( + "Record a delayed outcome for sources already marked used. This server uses the " + "outcome-driven deactivation candidate: confirmed follows the soft-start schedule; " + "causally attributed corrected and rolled_back outcomes exactly reverse each active " + "credited contribution together with its same-source sibling normalization mutations. " + "Superseded makes the saved relation path dormant without deleting evidence, and a " + "later confirmed outcome on that saved path reactivates it. Unattributed, duplicate, " + "lexical, and zero-hop outcomes remain non-mutating." +) _IDEMPOTENCY = re.compile(r"^[A-Za-z0-9._:-]+$") _TRACE = re.compile(r"^[0-9a-f]{32}$") @@ -198,6 +207,50 @@ def _object(properties: dict[str, Any], required: list[str]) -> dict[str, Any]: }, ["node_id", "steps"], ) +_DORMANCY_OUTPUT = _object( + { + "source_id": {"type": "string"}, + "target_id": {"type": "string"}, + "edge_type": {"type": "string"}, + "old_dormant": {"type": "boolean"}, + "new_dormant": {"type": "boolean"}, + }, + ["source_id", "target_id", "edge_type", "old_dormant", "new_dormant"], +) +_CONTRIBUTION_MUTATION_OUTPUT = _object( + { + "mutation_role": {"type": "string", "enum": ["credited", "sibling"]}, + "source_id": {"type": "string"}, + "target_id": {"type": "string"}, + "edge_type": {"type": "string"}, + "actual_delta": {"type": "number"}, + "old_weight": {"type": "number"}, + "new_weight": {"type": "number"}, + }, + [ + "mutation_role", "source_id", "target_id", "edge_type", + "actual_delta", "old_weight", "new_weight", + ], +) +_REVERSED_CONTRIBUTION_OUTPUT = _object( + { + "contribution_id": {"type": "string"}, + "contribution_kind": { + "type": "string", + "enum": ["soft_start_provisional", "soft_start_confirmation"], + }, + "source_record_id": {"type": "string"}, + "source_id": {"type": "string"}, + "target_id": {"type": "string"}, + "edge_type": {"type": "string"}, + "credited_delta": {"type": "number"}, + "mutations": {"type": "array", "items": _CONTRIBUTION_MUTATION_OUTPUT}, + }, + [ + "contribution_id", "contribution_kind", "source_record_id", "source_id", + "target_id", "edge_type", "credited_delta", "mutations", + ], +) SEARCH_OUTPUT = _object( { "contract_version": {"type": "string", "const": CONTRACT_VERSION}, @@ -317,6 +370,12 @@ def _object(properties: dict[str, Any], required: list[str]) -> dict[str, Any]: "confirmations": {"type": "array", "items": _CONFIRMATION_OUTPUT}, "credited_paths": {"type": "array", "items": _CREDITED_PATH_OUTPUT}, "normalized_sibling_edges": {"type": "array", "items": _EDGE_OUTPUT}, + "deactivation_applied": {"type": "boolean"}, + "reversed_contributions": { + "type": "array", "items": _REVERSED_CONTRIBUTION_OUTPUT + }, + "dormancy_changes": {"type": "array", "items": _DORMANCY_OUTPUT}, + "reactivated_edges": {"type": "array", "items": _DORMANCY_OUTPUT}, }, ["contract_version", "outcome_id", "trace_id", "node_ids", "outcome", "recorded_at", "reinforcement_applied"], ) @@ -360,6 +419,7 @@ def _tools( *, confirmed_outcome_reinforcement: bool, soft_start_feedback_reinforcement: bool, + outcome_driven_feedback_deactivation: bool = False, ) -> tuple[types.Tool, ...]: if not confirmed_outcome_reinforcement and not soft_start_feedback_reinforcement: return TOOLS @@ -369,9 +429,13 @@ def _tools( else CONFIRMED_SOURCE_USE_DESCRIPTION ) outcome_description = ( - SOFT_START_OUTCOME_DESCRIPTION - if soft_start_feedback_reinforcement - else CONFIRMED_OUTCOME_DESCRIPTION + DEACTIVATION_OUTCOME_DESCRIPTION + if outcome_driven_feedback_deactivation + else ( + SOFT_START_OUTCOME_DESCRIPTION + if soft_start_feedback_reinforcement + else CONFIRMED_OUTCOME_DESCRIPTION + ) ) return ( TOOLS[0], @@ -405,6 +469,9 @@ def __init__( soft_start_feedback_reinforcement=( self.engine.config.soft_start_feedback_reinforcement ), + outcome_driven_feedback_deactivation=( + self.engine.config.outcome_driven_feedback_deactivation + ), ) def close(self) -> None: @@ -646,6 +713,51 @@ def _record_outcome(self, data: dict[str, Any]) -> dict[str, Any]: } for edge in receipt.normalized_sibling_edges ], + "deactivation_applied": receipt.deactivation_applied, + "reversed_contributions": [ + { + "contribution_id": item.contribution_id, + "contribution_kind": item.contribution_kind, + "source_record_id": item.source_record_id, + "source_id": item.source_id, + "target_id": item.target_id, + "edge_type": item.edge_type, + "credited_delta": item.credited_delta, + "mutations": [ + { + "mutation_role": mutation.mutation_role, + "source_id": mutation.source_id, + "target_id": mutation.target_id, + "edge_type": mutation.edge_type, + "actual_delta": mutation.actual_delta, + "old_weight": mutation.old_weight, + "new_weight": mutation.new_weight, + } + for mutation in item.mutations + ], + } + for item in receipt.reversed_contributions + ], + "dormancy_changes": [ + { + "source_id": item.source_id, + "target_id": item.target_id, + "edge_type": item.edge_type, + "old_dormant": item.old_dormant, + "new_dormant": item.new_dormant, + } + for item in receipt.dormancy_changes + ], + "reactivated_edges": [ + { + "source_id": item.source_id, + "target_id": item.target_id, + "edge_type": item.edge_type, + "old_dormant": item.old_dormant, + "new_dormant": item.new_dormant, + } + for item in receipt.reactivated_edges + ], } @staticmethod @@ -843,6 +955,11 @@ def _build_parser() -> argparse.ArgumentParser: default=None, help="Provisional fraction required by soft-start feedback reinforcement", ) + parser.add_argument( + "--outcome-driven-feedback-deactivation", + action="store_true", + help="Exactly reverse attributed soft-start contributions or make them dormant", + ) return parser @@ -883,6 +1000,14 @@ def main() -> None: parser.error( "--soft-start-feedback-ratio requires --soft-start-feedback-reinforcement" ) + if ( + arguments.outcome_driven_feedback_deactivation + and not arguments.soft_start_feedback_reinforcement + ): + parser.error( + "--outcome-driven-feedback-deactivation requires " + "--soft-start-feedback-reinforcement" + ) config = EngineConfig( relation_feedback_evidence_quorum=( arguments.relation_feedback_evidence_quorum @@ -894,5 +1019,8 @@ def main() -> None: arguments.soft_start_feedback_reinforcement ), soft_start_feedback_ratio=arguments.soft_start_feedback_ratio, + outcome_driven_feedback_deactivation=( + arguments.outcome_driven_feedback_deactivation + ), ) asyncio.run(_run(arguments.database, config=config)) diff --git a/tests/fixtures/outcome_feedback_deactivation_v1.fixture.json b/tests/fixtures/outcome_feedback_deactivation_v1.fixture.json new file mode 100644 index 0000000..e22f536 --- /dev/null +++ b/tests/fixtures/outcome_feedback_deactivation_v1.fixture.json @@ -0,0 +1,130 @@ +{ + "protocol_id": "outcome-driven-feedback-deactivation-v1", + "selection_rule": "Fresh snapshot public identities and saved one-hop relation paths fixed before registered outputs; development and holdout identities are disjoint.", + "stages": { + "development": [ + { + "case_id": "deactivation-dev-corrected-requirements-corpus", + "case_role": "corrected", + "query": "requirements specification feedback interface", + "search_surface": "relation", + "used_node_id": "docs/d1-corpus-fixture.md", + "credited_edge": { + "source_id": "docs/requirements.md", + "target_id": "docs/d1-corpus-fixture.md", + "edge_type": "links_to" + }, + "registered_initial_state": { + "weight": 0.5, + "reinforced_count": 0 + }, + "negative_outcome": "corrected" + }, + { + "case_id": "deactivation-dev-rollback-readme-anchored", + "case_role": "rolled_back", + "query": "Neuron Graph RAG prototype", + "search_surface": "relation", + "used_node_id": "docs/anchored-bm25-graph-hybrid-experiment.md", + "credited_edge": { + "source_id": "README.md", + "target_id": "docs/anchored-bm25-graph-hybrid-experiment.md", + "edge_type": "links_to" + }, + "registered_initial_state": { + "weight": 0.5, + "reinforced_count": 0 + }, + "negative_outcome": "rolled_back" + }, + { + "case_id": "deactivation-dev-superseded-requirements-blind", + "case_role": "superseded", + "query": "requirements specification feedback interface", + "search_surface": "relation", + "used_node_id": "docs/blind-llm-channel-selection-experiment.md", + "credited_edge": { + "source_id": "docs/requirements.md", + "target_id": "docs/blind-llm-channel-selection-experiment.md", + "edge_type": "links_to" + }, + "registered_initial_state": { + "weight": 0.5, + "reinforced_count": 0 + }, + "negative_outcome": "superseded" + }, + { + "case_id": "deactivation-dev-unattributed-lexical-fusion", + "case_role": "unattributed", + "query": "anchored fusion calibration experiment", + "search_surface": "lexical", + "used_node_id": "docs/anchored-fusion-calibration-experiment.md", + "credited_edge": {}, + "negative_outcome": "corrected" + } + ], + "holdout": [ + { + "case_id": "deactivation-holdout-corrected-readme-node-first", + "case_role": "corrected", + "query": "Neuron Graph RAG prototype", + "search_surface": "relation", + "used_node_id": "docs/node-first-blind-selection-experiment.md", + "credited_edge": { + "source_id": "README.md", + "target_id": "docs/node-first-blind-selection-experiment.md", + "edge_type": "links_to" + }, + "registered_initial_state": { + "weight": 0.5, + "reinforced_count": 0 + }, + "negative_outcome": "corrected" + }, + { + "case_id": "deactivation-holdout-rollback-requirements-trajectory", + "case_role": "rolled_back", + "query": "requirements specification feedback interface", + "search_surface": "relation", + "used_node_id": "docs/engine-backed-feedback-trajectory-experiment.md", + "credited_edge": { + "source_id": "docs/requirements.md", + "target_id": "docs/engine-backed-feedback-trajectory-experiment.md", + "edge_type": "links_to" + }, + "registered_initial_state": { + "weight": 0.5, + "reinforced_count": 0 + }, + "negative_outcome": "rolled_back" + }, + { + "case_id": "deactivation-holdout-superseded-readme-benchmark", + "case_role": "superseded", + "query": "Neuron Graph RAG prototype", + "search_surface": "relation", + "used_node_id": "docs/real-corpus-benchmark.md", + "credited_edge": { + "source_id": "README.md", + "target_id": "docs/real-corpus-benchmark.md", + "edge_type": "links_to" + }, + "registered_initial_state": { + "weight": 0.5, + "reinforced_count": 0 + }, + "negative_outcome": "superseded" + }, + { + "case_id": "deactivation-holdout-unattributed-lexical-compatibility", + "case_role": "unattributed", + "query": "GitHub RAG MCP replacement compatibility", + "search_surface": "lexical", + "used_node_id": "docs/github-rag-mcp-replacement-compatibility.md", + "credited_edge": {}, + "negative_outcome": "rolled_back" + } + ] + } +} diff --git a/tests/fixtures/outcome_feedback_deactivation_v1.gate.json b/tests/fixtures/outcome_feedback_deactivation_v1.gate.json new file mode 100644 index 0000000..78d54f0 --- /dev/null +++ b/tests/fixtures/outcome_feedback_deactivation_v1.gate.json @@ -0,0 +1,37 @@ +{ + "protocol_id": "outcome-driven-feedback-deactivation-v1", + "gates": [ + { + "gate_id": "protocol-and-preflight-integrity", + "requirement": "Frozen hashes, snapshot, registered baseline, case identity, path identity, capacity, privacy, and output absence all pass." + }, + { + "gate_id": "corrected-exact-contribution-reversal", + "requirement": "Corrected reverses every active credited and same-contribution sibling mutation, while control remains audit-only." + }, + { + "gate_id": "rolled-back-exact-contribution-reversal", + "requirement": "Rolled back reverses the same persisted contribution unit after restart without touching another contribution." + }, + { + "gate_id": "superseded-dormancy-and-reactivation", + "requirement": "Superseded hides the stored edge from normal activation without deletion and confirmed on the saved path reactivates it." + }, + { + "gate_id": "baseline-floor-without-punitive-update", + "requirement": "No reversed credited edge falls below its registered baseline." + }, + { + "gate_id": "idempotency-restart-and-transaction-atomicity", + "requirement": "Replay, restart, and injected transaction failure cannot duplicate or partially apply deactivation." + }, + { + "gate_id": "credited-sibling-locality-and-controls", + "requirement": "Only attributed credited and same-source sibling mutations change; lexical and unattributed controls do not mutate." + }, + { + "gate_id": "source-snapshot-privacy-and-exclusive-output", + "requirement": "Source and snapshot remain unchanged, public output is private-free, and the registered output is exclusively created." + } + ] +} diff --git a/tests/fixtures/outcome_feedback_deactivation_v1.manifest.json b/tests/fixtures/outcome_feedback_deactivation_v1.manifest.json new file mode 100644 index 0000000..daf5fde --- /dev/null +++ b/tests/fixtures/outcome_feedback_deactivation_v1.manifest.json @@ -0,0 +1,72 @@ +{ + "protocol_id": "outcome-driven-feedback-deactivation-v1", + "issue": 111, + "source_boundary": "fresh-one-time-transaction-consistent-local-snapshot", + "registration_boundary": "freeze-only-squash-merge-before-successor-observation", + "snapshot": { + "source_locator": "local_codex_ngr_database", + "source_access": "sqlite-uri-mode-ro-query-only", + "capture_method": "sqlite-backup-api", + "captured_at": "2026-08-21T11:56:00+00:00", + "source_logical_sha256_before": "92fc59c96ec1bb70c846997e77d34f911d5cf98f65e68312ed22b86c891ef27a", + "source_logical_sha256_after": "92fc59c96ec1bb70c846997e77d34f911d5cf98f65e68312ed22b86c891ef27a", + "snapshot_logical_sha256": "92fc59c96ec1bb70c846997e77d34f911d5cf98f65e68312ed22b86c891ef27a", + "snapshot_sha256": "b129f1f85f2706929cdc6291c5c846b8a50d419cb3b49ca9f108987c6d2d2454", + "snapshot_size": 442368, + "schema_sha256": "b5cd9409af9d5107935ee857c7e84b31392ad428f18ea72ec861452ffa12867b", + "table_names": [ + "activation_state", + "confirmed_edge_state", + "confirmed_relation_feedback", + "confirmed_source_uses", + "delayed_outcome_nodes", + "delayed_outcomes", + "edges", + "feedback_requests", + "nodes", + "relation_feedback_evidence", + "retrieval_channels", + "retrieval_results", + "retrievals", + "soft_start_edge_state", + "soft_start_relation_feedback", + "source_use_state", + "success_feedback", + "success_nodes" + ], + "row_counts": { + "nodes": 21, + "edges": 30, + "retrievals": 24, + "success_feedback": 10, + "delayed_outcomes": 3 + } + }, + "protocol_artifacts": { + "fixture": "tests/fixtures/outcome_feedback_deactivation_v1.fixture.json", + "schedule": "tests/fixtures/outcome_feedback_deactivation_v1.schedule.json", + "gate": "tests/fixtures/outcome_feedback_deactivation_v1.gate.json", + "result_schema": "tests/fixtures/outcome_feedback_deactivation_v1.result-schema.json", + "audit": "tests/fixtures/outcome_feedback_deactivation_v1.result-free-audit.json" + }, + "outputs": { + "development": "tests/fixtures/outcome_feedback_deactivation_v1.development.observed.json", + "holdout": "tests/fixtures/outcome_feedback_deactivation_v1.holdout.observed.json" + }, + "artifact_sha256": { + "src/neuron_graph_rag/outcome_feedback_deactivation_evaluation.py": "ad3b7ffda22ec66b57734c5a2a367b1e9f8f39b397cf6da9c9cc244038214a57", + "tools/run_outcome_feedback_deactivation_evaluation.py": "ab41133fb4b97e1bd30d916fd7e1684d4c4d99a39580c5f32068c876da592ce8", + "tests/test_outcome_feedback_deactivation_evaluation.py": "52a75ae4b6542ef49be3bdc7d58d4a40cea800da04f2214bc458a45c82f9beda", + "tests/test_outcome_feedback_deactivation.py": "460407e0b5f97747cf6187b7e6dabcadac349930f6d511a2c9c4524923aef49d", + "tests/fixtures/outcome_feedback_deactivation_v1.fixture.json": "41446bf8f50da64dfcdafd60a72e036e61afb4ae2fa7d995cdfb049da4c822b2", + "tests/fixtures/outcome_feedback_deactivation_v1.schedule.json": "97b9d1df873d595addd7030c7b65caea4a0eda9e675cb5a1ecfe9ef7b062bd4a", + "tests/fixtures/outcome_feedback_deactivation_v1.gate.json": "3bbbfbb3a694bf7962694bac999d6b696786eeedb774612b2b7d0efcbbe1045b", + "tests/fixtures/outcome_feedback_deactivation_v1.result-schema.json": "f3a1b08f0725e5bac2231bc9dde5d7c9fb8d7a844c5860037e4157b5718581cb", + "tests/fixtures/outcome_feedback_deactivation_v1.result-free-audit.json": "ef27a7ba3c2a44c79387e891482b3b5ea8a100172f23c8932aad3a2099b8d2d5", + "docs/outcome-driven-feedback-deactivation.md": "27ebe4c66151bfb78ba03e687961d3eb75d163a4bc843662686712ad5e0d517d", + "docs/requirements.md": "1036da4d8adb3293dc6ce8cf0894778bea63dfccba7a94f3a10f97f869107cc0", + "docs/Decision-Structure.md": "bf074a79e838cff67a5560a2fa82278f983dd01b1dd04bad50afe621217e525c", + "docs/optional-mcp-interface.md": "75dc2fdcc769e22fbc6b075ecfe5306935e05f30a52d1718a6d47e930f3d7102", + "README.md": "d65962bd1b5564e87a9da5bc9ccddcd864c64c259dc6a408816ca852eecd35c1" + } +} diff --git a/tests/fixtures/outcome_feedback_deactivation_v1.result-free-audit.json b/tests/fixtures/outcome_feedback_deactivation_v1.result-free-audit.json new file mode 100644 index 0000000..f30d9f6 --- /dev/null +++ b/tests/fixtures/outcome_feedback_deactivation_v1.result-free-audit.json @@ -0,0 +1,13 @@ +{ + "protocol_id": "outcome-driven-feedback-deactivation-v1", + "registered_development_output_present": false, + "registered_holdout_output_present": false, + "development_executed_in_freeze_issue": false, + "holdout_executed_in_freeze_issue": false, + "placeholder_round_trip_passed": true, + "source_database_mutated": false, + "live_config_mutated": false, + "private_snapshot_committed": false, + "private_body_committed": false, + "freeze_observation_split": "Observation begins only in a successor issue after the squash-merged freeze registration is present on main." +} diff --git a/tests/fixtures/outcome_feedback_deactivation_v1.result-schema.json b/tests/fixtures/outcome_feedback_deactivation_v1.result-schema.json new file mode 100644 index 0000000..936bce3 --- /dev/null +++ b/tests/fixtures/outcome_feedback_deactivation_v1.result-schema.json @@ -0,0 +1,21 @@ +{ + "protocol_id": "outcome-driven-feedback-deactivation-v1", + "required": [ + "protocol_id", + "stage", + "snapshot_sha256", + "arms", + "hard_gates", + "all_hard_gates_pass" + ], + "stage_enum": [ + "development", + "holdout" + ], + "arm_order": [ + "control", + "deactivation_candidate" + ], + "writer": "exclusive-create-no-overwrite", + "verifier": "canonical-json-identity-gate-order-and-aggregate" +} diff --git a/tests/fixtures/outcome_feedback_deactivation_v1.schedule.json b/tests/fixtures/outcome_feedback_deactivation_v1.schedule.json new file mode 100644 index 0000000..5bb3d94 --- /dev/null +++ b/tests/fixtures/outcome_feedback_deactivation_v1.schedule.json @@ -0,0 +1,52 @@ +{ + "protocol_id": "outcome-driven-feedback-deactivation-v1", + "arms": [ + { + "arm_id": "control", + "soft_start_feedback_reinforcement": true, + "soft_start_feedback_ratio": 0.25, + "confirmation_decay_ratio": 0.5, + "sibling_feedback_normalization": 1.0, + "outcome_driven_feedback_deactivation": false + }, + { + "arm_id": "deactivation_candidate", + "soft_start_feedback_reinforcement": true, + "soft_start_feedback_ratio": 0.25, + "confirmation_decay_ratio": 0.5, + "sibling_feedback_normalization": 1.0, + "outcome_driven_feedback_deactivation": true + } + ], + "event_order": { + "corrected": [ + "relation_search", + "used", + "confirmed", + "corrected", + "idempotency_replay" + ], + "rolled_back": [ + "relation_search", + "used", + "confirmed", + "restart", + "rolled_back", + "idempotency_replay" + ], + "superseded": [ + "relation_search", + "used", + "superseded", + "dormancy_checkpoint", + "confirmed_reactivation" + ], + "unattributed": [ + "lexical_search", + "used", + "negative_outcome", + "idempotency_replay" + ] + }, + "execution": "fresh snapshot clone per case and arm; development once; holdout once only after all development hard gates pass" +} diff --git a/tests/test_mcp_adapter.py b/tests/test_mcp_adapter.py index de6daca..4072ebf 100644 --- a/tests/test_mcp_adapter.py +++ b/tests/test_mcp_adapter.py @@ -18,6 +18,7 @@ CONFIRMED_OUTCOME_DESCRIPTION, CONFIRMED_SOURCE_USE_DESCRIPTION, CONTRACT_VERSION, + DEACTIVATION_OUTCOME_DESCRIPTION, OUTCOME_DESCRIPTION, SEARCH_DESCRIPTION, SOFT_START_OUTCOME_DESCRIPTION, @@ -380,6 +381,95 @@ async def test_soft_start_candidate_receipt_parity_and_provenance(self) -> None: confirmation["actual_delta"], ) + async def test_deactivation_candidate_receipt_description_and_provenance(self) -> None: + self.adapter.close() + self.adapter = FeedbackMCPAdapter( + self.database, + config=EngineConfig( + soft_start_feedback_reinforcement=True, + soft_start_feedback_ratio=0.25, + confirmation_decay_ratio=0.5, + sibling_feedback_normalization=1.0, + outcome_driven_feedback_deactivation=True, + ), + ) + listed = await self.adapter.list_tools() + self.assertEqual(listed.tools[2].description, DEACTIVATION_OUTCOME_DESCRIPTION) + search = await self.adapter.call_tool( + None, + types.CallToolRequestParams( + name="search", + arguments={ + "contract_version": CONTRACT_VERSION, + "query": "cache invalidation", + "limit": 3, + }, + ), + ) + self.assertTrue( + search.structured_content["effective_config_provenance"] + ["effective_config"]["feedback"] + ["outcome_driven_feedback_deactivation"] + ) + trace_id = search.structured_content["trace_id"] + await self.adapter.call_tool( + None, + types.CallToolRequestParams( + name="record_source_use", + arguments={ + "contract_version": CONTRACT_VERSION, + "idempotency_key": "deactivation-mcp-use", + "trace_id": trace_id, + "events": [ + {"node_id": "implementation", "stage": "selected"}, + {"node_id": "implementation", "stage": "validated"}, + {"node_id": "implementation", "stage": "used"}, + ], + }, + ), + ) + await self.adapter.call_tool( + None, + types.CallToolRequestParams( + name="record_outcome", + arguments={ + "contract_version": CONTRACT_VERSION, + "idempotency_key": "deactivation-mcp-confirmed", + "trace_id": trace_id, + "node_ids": ["implementation"], + "outcome": "confirmed", + "summary": "confirmed before correction", + }, + ), + ) + corrected = await self.adapter.call_tool( + None, + types.CallToolRequestParams( + name="record_outcome", + arguments={ + "contract_version": CONTRACT_VERSION, + "idempotency_key": "deactivation-mcp-corrected", + "trace_id": trace_id, + "node_ids": ["implementation"], + "outcome": "corrected", + "summary": "credited result was corrected", + }, + ), + ) + payload = corrected.structured_content + self.assertTrue(payload["deactivation_applied"]) + self.assertEqual(len(payload["reversed_contributions"]), 2) + self.assertEqual(payload["dormancy_changes"], []) + self.assertEqual(payload["reactivated_edges"], []) + self.assertEqual( + { + mutation["mutation_role"] + for contribution in payload["reversed_contributions"] + for mutation in contribution["mutations"] + }, + {"credited", "sibling"}, + ) + async def test_stdio_protocol_smoke(self) -> None: parameters = StdioServerParameters( command=sys.executable, @@ -577,6 +667,7 @@ def test_invalid_cli_feedback_settings_do_not_create_database(self) -> None: invalid_combinations = ( ("--soft-start-feedback-reinforcement", "--confirmation-decay-ratio", "0.5"), ("--soft-start-feedback-ratio", "0.25"), + ("--outcome-driven-feedback-deactivation",), ( "--confirmed-outcome-reinforcement", "--soft-start-feedback-reinforcement", diff --git a/tests/test_outcome_feedback_deactivation.py b/tests/test_outcome_feedback_deactivation.py new file mode 100644 index 0000000..edd40b6 --- /dev/null +++ b/tests/test_outcome_feedback_deactivation.py @@ -0,0 +1,326 @@ +from __future__ import annotations + +import sqlite3 +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from neuron_graph_rag import FeedbackLedger, SourceUseEvent +from neuron_graph_rag.config_provenance import effective_config_provenance +from neuron_graph_rag.evidence_feedback import EngineConfig, NeuronGraphRAG + + +def _config() -> EngineConfig: + return EngineConfig( + sparse_weight=1.0, + dense_weight=0.0, + seed_count=1, + max_hops=2, + feedback_learning_rate=0.2, + sibling_feedback_normalization=1.0, + soft_start_feedback_reinforcement=True, + soft_start_feedback_ratio=0.25, + confirmation_decay_ratio=0.5, + outcome_driven_feedback_deactivation=True, + ) + + +def _populate(engine: NeuronGraphRAG) -> None: + engine.add_document("source", "alpha lexical source") + engine.add_document("target", "distant relation target") + engine.add_document("sibling", "uncredited sibling") + engine.add_document("other-source", "isolated origin") + engine.add_document("other-target", "isolated destination") + engine.add_edge("source", "target", "supports", weight=0.5) + engine.add_edge("source", "sibling", "supports", weight=0.4) + engine.add_edge("other-source", "other-target", "isolated", weight=0.8) + + +def _used_relation( + engine: NeuronGraphRAG, index: int +) -> tuple[FeedbackLedger, str, object]: + ledger = FeedbackLedger(engine) + trace = engine.search_channels("alpha", limit=5, now=1_000.0 + index).relation + receipt = ledger.record_source_use( + trace.trace_id, + [ + SourceUseEvent("target", "selected"), + SourceUseEvent("target", "validated"), + SourceUseEvent("target", "used"), + ], + idempotency_key=f"deactivation-use-{index}", + now=2_000.0 + index, + ) + return ledger, trace.trace_id, receipt + + +class OutcomeFeedbackDeactivationTest(unittest.TestCase): + def test_config_is_default_off_and_requires_soft_start(self) -> None: + default = EngineConfig() + self.assertFalse(default.outcome_driven_feedback_deactivation) + feedback = effective_config_provenance(default)["effective_config"]["feedback"] + self.assertNotIn("outcome_driven_feedback_deactivation", feedback) + active = effective_config_provenance(_config())["effective_config"]["feedback"] + self.assertTrue(active["outcome_driven_feedback_deactivation"]) + with self.assertRaises(ValueError): + EngineConfig(outcome_driven_feedback_deactivation=True) + + def test_correction_exactly_reverses_credited_and_sibling_mutations(self) -> None: + with NeuronGraphRAG(config=_config()) as engine: + _populate(engine) + ledger, trace_id, used = _used_relation(engine, 1) + self.assertIsNotNone(used.feedback) + confirmed = ledger.record_outcome( + trace_id, + ["target"], + "confirmed", + "confirmed first", + idempotency_key="deactivation-confirmed", + now=3_000.0, + ) + self.assertTrue(confirmed.reinforcement_applied) + self.assertGreater( + engine.store.edge("source", "target", "supports").weight, 0.5 + ) + self.assertLess( + engine.store.edge("source", "sibling", "supports").weight, 0.4 + ) + + corrected = ledger.record_outcome( + trace_id, + ["target"], + "corrected", + "credited answer was wrong", + idempotency_key="deactivation-corrected", + now=4_000.0, + ) + self.assertTrue(corrected.deactivation_applied) + self.assertEqual(len(corrected.reversed_contributions), 2) + self.assertEqual( + {item.contribution_kind for item in corrected.reversed_contributions}, + {"soft_start_provisional", "soft_start_confirmation"}, + ) + confirmation = next( + item + for item in corrected.reversed_contributions + if item.contribution_kind == "soft_start_confirmation" + ) + self.assertEqual( + {mutation.mutation_role for mutation in confirmation.mutations}, + {"credited", "sibling"}, + ) + self.assertAlmostEqual( + engine.store.edge("source", "target", "supports").weight, 0.5 + ) + self.assertAlmostEqual( + engine.store.edge("source", "sibling", "supports").weight, 0.4 + ) + self.assertAlmostEqual( + engine.store.edge("other-source", "other-target", "isolated").weight, + 0.8, + ) + + replay = ledger.record_outcome( + trace_id, + ["target"], + "corrected", + "credited answer was wrong", + idempotency_key="deactivation-corrected", + now=9_000.0, + ) + self.assertEqual(replay, corrected) + self.assertAlmostEqual( + engine.store.edge("source", "target", "supports").weight, 0.5 + ) + + def test_rollback_is_trace_scoped_and_persists_across_restart(self) -> None: + with tempfile.TemporaryDirectory() as directory: + database = Path(directory) / "deactivation.sqlite" + with NeuronGraphRAG(database, config=_config()) as engine: + _populate(engine) + ledger, first_trace, _ = _used_relation(engine, 1) + _, second_trace, _ = _used_relation(engine, 2) + ledger.record_outcome( + second_trace, + ["target"], + "confirmed", + "second trace remains active", + idempotency_key="deactivation-second-confirmed", + ) + before_rollback = engine.store.edge( + "source", "target", "supports" + ).weight + first_delta = engine.store.connection.execute( + "SELECT credited_delta FROM feedback_contributions " + "WHERE trace_id = ? AND active = 1", + (first_trace,), + ).fetchone()[0] + with NeuronGraphRAG(database, config=_config()) as engine: + receipt = FeedbackLedger(engine).record_outcome( + first_trace, + ["target"], + "rolled_back", + "first use was rolled back", + idempotency_key="deactivation-rollback", + ) + self.assertTrue(receipt.deactivation_applied) + self.assertEqual(len(receipt.reversed_contributions), 1) + self.assertAlmostEqual( + engine.store.edge("source", "target", "supports").weight, + before_rollback - first_delta, + ) + active_second = engine.store.connection.execute( + "SELECT COUNT(*) FROM feedback_contributions " + "WHERE trace_id = ? AND active = 1", + (second_trace,), + ).fetchone()[0] + self.assertEqual(active_second, 1) + + def test_superseded_edge_is_dormant_until_saved_path_is_confirmed(self) -> None: + with NeuronGraphRAG(config=_config()) as engine: + _populate(engine) + ledger, trace_id, _ = _used_relation(engine, 1) + superseded = ledger.record_outcome( + trace_id, + ["target"], + "superseded", + "relationship became stale", + idempotency_key="deactivation-superseded", + ) + self.assertTrue(superseded.deactivation_applied) + self.assertTrue(engine.store.edge_is_dormant("source", "target", "supports")) + self.assertNotIn( + "target", + { + edge.target_id + for edge in engine.store.outgoing_edges("source") + }, + ) + + confirmed = ledger.record_outcome( + trace_id, + ["target"], + "confirmed", + "saved path is current again", + idempotency_key="deactivation-reactivated", + ) + self.assertEqual(len(confirmed.reactivated_edges), 1) + self.assertFalse(engine.store.edge_is_dormant("source", "target", "supports")) + self.assertIn( + "target", + { + edge.target_id + for edge in engine.store.outgoing_edges("source") + }, + ) + + def test_duplicate_confirmation_does_not_reactivate_and_reversal_has_floor(self) -> None: + with NeuronGraphRAG(config=_config()) as engine: + _populate(engine) + ledger, first_trace, _ = _used_relation(engine, 1) + ledger.record_outcome( + first_trace, + ["target"], + "confirmed", + "first independent confirmation", + idempotency_key="deactivation-floor-confirmed", + ) + _, saved_trace, _ = _used_relation(engine, 2) + ledger.record_outcome( + first_trace, + ["target"], + "superseded", + "relationship became stale", + idempotency_key="deactivation-floor-superseded", + ) + duplicate_trace = ledger.record_outcome( + first_trace, + ["target"], + "confirmed", + "duplicate trace is not new evidence", + idempotency_key="deactivation-floor-duplicate-confirmed", + ) + self.assertEqual(duplicate_trace.reactivated_edges, ()) + self.assertTrue(engine.store.edge_is_dormant("source", "target", "supports")) + independent = ledger.record_outcome( + saved_trace, + ["target"], + "confirmed", + "independent saved path is current", + idempotency_key="deactivation-floor-independent-confirmed", + ) + self.assertEqual(len(independent.reactivated_edges), 1) + + engine.store.connection.execute( + "UPDATE edges SET weight = 0.49 " + "WHERE source_id = 'source' AND target_id = 'target' " + "AND edge_type = 'supports'" + ) + engine.store.connection.commit() + corrected = ledger.record_outcome( + first_trace, + ["target"], + "corrected", + "reverse without punitive floor crossing", + idempotency_key="deactivation-floor-corrected", + ) + self.assertTrue(corrected.deactivation_applied) + self.assertGreaterEqual( + engine.store.edge("source", "target", "supports").weight, 0.5 + ) + + def test_unattributed_and_atomic_failures_do_not_partially_mutate(self) -> None: + with NeuronGraphRAG(config=_config()) as engine: + _populate(engine) + ledger, trace_id, _ = _used_relation(engine, 1) + before = engine.store.list_edges() + lexical = engine.search("alpha", limit=5, now=5_000.0) + ledger.record_source_use( + lexical.trace_id, + [ + SourceUseEvent("source", "selected"), + SourceUseEvent("source", "validated"), + SourceUseEvent("source", "used"), + ], + idempotency_key="deactivation-lexical-use", + ) + unattributed = ledger.record_outcome( + lexical.trace_id, + ["source"], + "corrected", + "lexical result", + idempotency_key="deactivation-lexical-correction", + ) + self.assertFalse(unattributed.deactivation_applied) + self.assertEqual(engine.store.list_edges(), before) + + with ( + patch.object( + engine.store, + "_save_idempotent_result", + side_effect=RuntimeError("injected deactivation receipt failure"), + ), + self.assertRaisesRegex( + RuntimeError, "injected deactivation receipt failure" + ), + ): + ledger.record_outcome( + trace_id, + ["target"], + "corrected", + "must roll back atomically", + idempotency_key="deactivation-atomic", + ) + self.assertEqual(engine.store.list_edges(), before) + self.assertEqual( + engine.store.connection.execute( + "SELECT COUNT(*) FROM delayed_outcomes " + "WHERE summary = 'must roll back atomically'" + ).fetchone()[0], + 0, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_outcome_feedback_deactivation_evaluation.py b/tests/test_outcome_feedback_deactivation_evaluation.py new file mode 100644 index 0000000..090effc --- /dev/null +++ b/tests/test_outcome_feedback_deactivation_evaluation.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import hashlib +import json +import sqlite3 +import subprocess +import tempfile +import unittest +from pathlib import Path + +from neuron_graph_rag.corpus_integrity import verify_manifest_source_hashes +from neuron_graph_rag.outcome_feedback_deactivation_evaluation import ( + ARMS, + MANIFEST_PATH, + PROTOCOL_ID, + STEM, + acquire_transactional_snapshot, + assert_public_payload, + prove_writer_verifier_round_trip, + read_json, + write_json_exclusive, +) + + +ROOT = Path(__file__).resolve().parents[1] +FIXTURES = ROOT / "tests" / "fixtures" + + +def _fixture(name: str) -> dict[str, object]: + return read_json(FIXTURES / f"{STEM}.{name}.json") + + +class OutcomeFeedbackDeactivationFreezeTest(unittest.TestCase): + def test_result_free_artifacts_are_canonical_private_free_and_complete(self) -> None: + for name in ( + "fixture", + "schedule", + "gate", + "result-schema", + "result-free-audit", + ): + path = FIXTURES / f"{STEM}.{name}.json" + raw = path.read_bytes() + payload = json.loads(raw.decode("utf-8", errors="strict")) + self.assertNotIn(b"\r", raw) + self.assertEqual( + raw.decode("utf-8"), + json.dumps(payload, ensure_ascii=False, indent=2) + "\n", + ) + self.assertEqual(payload["protocol_id"], PROTOCOL_ID) + assert_public_payload(payload) + + schedule = _fixture("schedule") + self.assertEqual([arm["arm_id"] for arm in schedule["arms"]], list(ARMS)) + gate = _fixture("gate") + self.assertEqual(len(gate["gates"]), 8) + self.assertNotEqual( + [item["gate_id"] for item in gate["gates"]], + sorted(item["gate_id"] for item in gate["gates"]), + ) + audit = _fixture("result-free-audit") + self.assertFalse(audit["development_executed_in_freeze_issue"]) + self.assertFalse(audit["holdout_executed_in_freeze_issue"]) + + def test_case_identity_roles_and_paths_are_frozen(self) -> None: + fixture = _fixture("fixture") + seen: set[str] = set() + stage_nodes: dict[str, set[str]] = {} + for stage in ("development", "holdout"): + cases = fixture["stages"][stage] + self.assertEqual( + {case["case_role"] for case in cases}, + {"corrected", "rolled_back", "superseded", "unattributed"}, + ) + ids = {case["case_id"] for case in cases} + self.assertTrue(ids.isdisjoint(seen)) + seen.update(ids) + stage_nodes[stage] = {case["used_node_id"] for case in cases} + for case in cases: + if case["search_surface"] == "relation": + self.assertEqual( + case["credited_edge"]["target_id"], case["used_node_id"] + ) + self.assertEqual(case["registered_initial_state"]["weight"], 0.5) + self.assertTrue(stage_nodes["development"].isdisjoint(stage_nodes["holdout"])) + + def test_manifest_uses_registration_bytes_and_outputs_were_absent(self) -> None: + manifest = read_json(MANIFEST_PATH) + registered = verify_manifest_source_hashes( + ROOT, MANIFEST_PATH, manifest["artifact_sha256"] + ) + self.assertEqual(registered.artifact_sha256, manifest["artifact_sha256"]) + for relative in manifest["outputs"].values(): + completed = subprocess.run( + ["git", "cat-file", "-e", f"{registered.source_commit}:{relative}"], + cwd=ROOT, + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + self.assertNotEqual(completed.returncode, 0) + + def test_snapshot_acquisition_is_logically_stable_private_and_exclusive(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "source.sqlite" + snapshot = root / "snapshot.sqlite" + connection = sqlite3.connect(source) + try: + connection.executescript( + """ + CREATE TABLE nodes(node_id TEXT PRIMARY KEY, text TEXT); + CREATE TABLE edges(source_id TEXT, target_id TEXT, edge_type TEXT); + CREATE TABLE retrievals(trace_id TEXT); + CREATE TABLE success_feedback(feedback_id TEXT); + CREATE TABLE delayed_outcomes(outcome_id TEXT); + INSERT INTO nodes VALUES ('public-id', 'private body'); + """ + ) + connection.commit() + finally: + connection.close() + provenance = acquire_transactional_snapshot(source, snapshot) + self.assertEqual( + provenance["source_logical_sha256_before"], + provenance["source_logical_sha256_after"], + ) + self.assertEqual( + provenance["snapshot_sha256"], + hashlib.sha256(snapshot.read_bytes()).hexdigest(), + ) + self.assertNotIn("private body", json.dumps(provenance)) + assert_public_payload(provenance) + with self.assertRaises(FileExistsError): + acquire_transactional_snapshot(source, snapshot) + + def test_writer_round_trip_privacy_and_exclusive_creation(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + prove_writer_verifier_round_trip(root / "placeholder.json") + self.assertFalse((root / "placeholder.json").exists()) + target = root / "exclusive.json" + write_json_exclusive(target, {"safe": True}) + with self.assertRaises(FileExistsError): + write_json_exclusive(target, {"safe": False}) + for payload in ( + {"node_text": "private body"}, + {"value": "C:\\private\\snapshot.sqlite"}, + {"value": "/private/snapshot.sqlite"}, + {"value": "password=hunter2"}, + ): + with self.assertRaises(ValueError): + assert_public_payload(payload) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/run_outcome_feedback_deactivation_evaluation.py b/tools/run_outcome_feedback_deactivation_evaluation.py new file mode 100644 index 0000000..f10d8fc --- /dev/null +++ b/tools/run_outcome_feedback_deactivation_evaluation.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from tempfile import TemporaryDirectory + +from neuron_graph_rag.outcome_feedback_deactivation_evaluation import ( + acquire_transactional_snapshot, + preflight_snapshot, + prove_writer_verifier_round_trip, + run_registered_stage, + verify_registered_result, +) + + +def main() -> None: + parser = argparse.ArgumentParser() + actions = parser.add_mutually_exclusive_group(required=True) + actions.add_argument("--acquire", action="store_true") + actions.add_argument("--probe", action="store_true") + actions.add_argument("--stage", choices=("development", "holdout")) + actions.add_argument("--verify", choices=("development", "holdout")) + parser.add_argument("--source", type=Path) + parser.add_argument("--snapshot", type=Path) + args = parser.parse_args() + if args.acquire: + if args.source is None or args.snapshot is None: + parser.error("--acquire requires --source and --snapshot") + print(json.dumps(acquire_transactional_snapshot(args.source, args.snapshot), indent=2)) + elif args.probe: + if args.snapshot is None: + parser.error("--probe requires --snapshot") + with TemporaryDirectory() as directory: + prove_writer_verifier_round_trip(Path(directory) / "placeholder.json") + print(json.dumps(preflight_snapshot(args.snapshot), indent=2)) + elif args.stage: + if args.snapshot is None: + parser.error("--stage requires --snapshot") + print(run_registered_stage(args.stage, args.snapshot)) + else: + verify_registered_result(args.verify) + print(f"{args.verify} verification passed") + + +if __name__ == "__main__": + main() From 15338b5e906154724cd397dab6693fd27243071f Mon Sep 17 00:00:00 2001 From: lipluscodex <268560960+lipluscodex@users.noreply.github.com> Date: Fri, 21 Aug 2026 21:34:52 +0900 Subject: [PATCH 2/2] fix: preserve bounded interleaved reversal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 判定: - Accept [P1]: sibling normalization の逆適用後に、後続の credited contribution と合成された edge が maximum_edge_weight を超える問題を受理した。単純な上限 clip だけでは後続 contribution の逆適用時に切り詰め量を復元できず、exact reversal を失うため。 - Reject: なし。 修正: - edge ごとに journal 開始時の weight、reinforced count、上限を保存し、active な signed mutation の総和から逆適用後の状態を再構成する。 - near-ceiling sibling mutation と後続 credited contribution の交錯順序を固定した独立回帰テストを追加する。 - frozen manifest、protocol、gate、audit、既存 mechanics assertion の bytes は変更しない。 --- src/neuron_graph_rag/storage.py | 243 ++++++++++++++---- ...come_feedback_deactivation_interleaving.py | 100 +++++++ 2 files changed, 287 insertions(+), 56 deletions(-) create mode 100644 tests/test_outcome_feedback_deactivation_interleaving.py diff --git a/src/neuron_graph_rag/storage.py b/src/neuron_graph_rag/storage.py index 5b4b076..a44c6e2 100644 --- a/src/neuron_graph_rag/storage.py +++ b/src/neuron_graph_rag/storage.py @@ -254,6 +254,19 @@ def _create_schema(self) -> None: REFERENCES edges(source_id, target_id, edge_type) ON DELETE CASCADE ); + CREATE TABLE IF NOT EXISTS feedback_edge_journal_state ( + source_id TEXT NOT NULL, + target_id TEXT NOT NULL, + edge_type TEXT NOT NULL, + baseline_weight REAL NOT NULL CHECK(baseline_weight >= 0.0), + baseline_reinforced_count INTEGER NOT NULL + CHECK(baseline_reinforced_count >= 0), + maximum_weight REAL NOT NULL CHECK(maximum_weight >= baseline_weight), + PRIMARY KEY (source_id, target_id, edge_type), + FOREIGN KEY (source_id, target_id, edge_type) + REFERENCES edges(source_id, target_id, edge_type) ON DELETE CASCADE + ); + CREATE TABLE IF NOT EXISTS relation_edge_dormancy ( source_id TEXT NOT NULL, target_id TEXT NOT NULL, @@ -380,6 +393,35 @@ def _contribution_id( ) return hashlib.sha256(value.encode("utf-8")).hexdigest() + @classmethod + def _ensure_edge_journal_state( + cls, + connection: sqlite3.Connection, + *, + source_id: str, + target_id: str, + edge_type: str, + baseline_weight: float, + baseline_reinforced_count: int, + maximum_weight: float, + ) -> None: + connection.execute( + """ + INSERT OR IGNORE INTO feedback_edge_journal_state( + source_id, target_id, edge_type, baseline_weight, + baseline_reinforced_count, maximum_weight + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + source_id, + target_id, + edge_type, + baseline_weight, + baseline_reinforced_count, + max(baseline_weight, maximum_weight), + ), + ) + @classmethod def _insert_contribution( cls, @@ -392,15 +434,29 @@ def _insert_contribution( target_id: str, edge_type: str, baseline_weight: float, + edge_weight_before: float, + edge_reinforced_count_before: int, + maximum_weight: float, credited_delta: float, created_at: float, - sibling_mutations: Iterable[tuple[str, str, str, float]] = (), + sibling_mutations: Iterable[ + tuple[str, str, str, float, float, int] + ] = (), ) -> str | None: if credited_delta <= 0.0: return None contribution_id = cls._contribution_id( contribution_kind, source_record_id, source_id, target_id, edge_type ) + cls._ensure_edge_journal_state( + connection, + source_id=source_id, + target_id=target_id, + edge_type=edge_type, + baseline_weight=edge_weight_before, + baseline_reinforced_count=edge_reinforced_count_before, + maximum_weight=maximum_weight, + ) connection.execute( """ INSERT INTO feedback_contributions( @@ -431,9 +487,25 @@ def _insert_contribution( """, (contribution_id, source_id, target_id, edge_type, credited_delta), ) - for sibling_source, sibling_target, sibling_type, reduction in sibling_mutations: + for ( + sibling_source, + sibling_target, + sibling_type, + reduction, + sibling_old_weight, + sibling_reinforced_count, + ) in sibling_mutations: if reduction <= 0.0: continue + cls._ensure_edge_journal_state( + connection, + source_id=sibling_source, + target_id=sibling_target, + edge_type=sibling_type, + baseline_weight=sibling_old_weight, + baseline_reinforced_count=sibling_reinforced_count, + maximum_weight=maximum_weight, + ) connection.execute( """ INSERT INTO feedback_contribution_mutations( @@ -451,6 +523,63 @@ def _insert_contribution( ) return contribution_id + @staticmethod + def _rebuild_edge_from_active_journal( + connection: sqlite3.Connection, + source_id: str, + target_id: str, + edge_type: str, + ) -> tuple[float, int]: + state = connection.execute( + """ + SELECT * FROM feedback_edge_journal_state + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + (source_id, target_id, edge_type), + ).fetchone() + if state is None: + raise KeyError("journaled edge state is absent") + active_delta = float( + connection.execute( + """ + SELECT COALESCE(SUM(mutation.actual_delta), 0.0) + FROM feedback_contribution_mutations AS mutation + JOIN feedback_contributions AS contribution + ON contribution.contribution_id = mutation.contribution_id + WHERE mutation.source_id = ? AND mutation.target_id = ? + AND mutation.edge_type = ? AND contribution.active = 1 + """, + (source_id, target_id, edge_type), + ).fetchone()[0] + ) + active_reinforced_count = int( + connection.execute( + """ + SELECT COALESCE(SUM(contribution.reinforced_count_delta), 0) + FROM feedback_contribution_mutations AS mutation + JOIN feedback_contributions AS contribution + ON contribution.contribution_id = mutation.contribution_id + WHERE mutation.source_id = ? AND mutation.target_id = ? + AND mutation.edge_type = ? AND mutation.mutation_role = 'credited' + AND contribution.active = 1 + """, + (source_id, target_id, edge_type), + ).fetchone()[0] + ) + new_weight = min( + float(state["maximum_weight"]), + max(0.0, float(state["baseline_weight"]) + active_delta), + ) + new_count = int(state["baseline_reinforced_count"]) + active_reinforced_count + connection.execute( + """ + UPDATE edges SET weight = ?, reinforced_count = ? + WHERE source_id = ? AND target_id = ? AND edge_type = ? + """, + (new_weight, new_count, source_id, target_id, edge_type), + ) + return new_weight, new_count + def edge(self, source_id: str, target_id: str, edge_type: str) -> TypedEdge: row = self.connection.execute( """ @@ -934,7 +1063,7 @@ def apply_soft_start_feedback( continue edge = connection.execute( """ - SELECT weight FROM edges + SELECT weight, reinforced_count FROM edges WHERE source_id = ? AND target_id = ? AND edge_type = ? """, (source_id, target_id, edge_type), @@ -1055,6 +1184,9 @@ def apply_soft_start_feedback( baseline_weight=( old_weight if state is None else float(state["initial_weight"]) ), + edge_weight_before=old_weight, + edge_reinforced_count_before=int(edge["reinforced_count"]), + maximum_weight=maximum, credited_delta=actual_delta, created_at=created_at, ) @@ -1443,7 +1575,7 @@ def record_soft_start_confirmed_outcome( normalized: list[dict[str, Any]] = [] contribution_specs: list[dict[str, Any]] = [] sibling_reductions_by_source: dict[ - str, list[tuple[str, str, str, float]] + str, list[tuple[str, str, str, float, float, int]] ] = {} reactivated: list[dict[str, Any]] = [] with self.transaction() as connection: @@ -1513,7 +1645,7 @@ def record_soft_start_confirmed_outcome( ) edge = connection.execute( """ - SELECT weight FROM edges + SELECT weight, reinforced_count FROM edges WHERE source_id = ? AND target_id = ? AND edge_type = ? """, (source_id, target_id, edge_type), @@ -1628,6 +1760,11 @@ def record_soft_start_confirmed_outcome( "target_id": target_id, "edge_type": edge_type, "baseline_weight": float(state["initial_weight"]), + "edge_weight_before": old_weight, + "edge_reinforced_count_before": int( + edge["reinforced_count"] + ), + "maximum_weight": maximum, "credited_delta": actual_delta, } ) @@ -1643,7 +1780,7 @@ def record_soft_start_confirmed_outcome( for sibling_source, target_id, edge_type in sibling_keys: row = connection.execute( """ - SELECT weight FROM edges + SELECT weight, reinforced_count FROM edges WHERE source_id = ? AND target_id = ? AND edge_type = ? """, (sibling_source, target_id, edge_type), @@ -1677,6 +1814,8 @@ def record_soft_start_confirmed_outcome( target_id, edge_type, old_weight - new_weight, + old_weight, + int(row["reinforced_count"]), ) ) for spec in contribution_specs: @@ -1692,6 +1831,11 @@ def record_soft_start_confirmed_outcome( target_id=str(spec["target_id"]), edge_type=str(spec["edge_type"]), baseline_weight=float(spec["baseline_weight"]), + edge_weight_before=float(spec["edge_weight_before"]), + edge_reinforced_count_before=int( + spec["edge_reinforced_count_before"] + ), + maximum_weight=float(spec["maximum_weight"]), credited_delta=float(spec["credited_delta"]), created_at=recorded_at, sibling_mutations=tuple( @@ -1700,8 +1844,17 @@ def record_soft_start_confirmed_outcome( sibling_target, sibling_type, reduction * share, + sibling_old_weight, + sibling_reinforced_count, + ) + for ( + sibling_source, + sibling_target, + sibling_type, + reduction, + sibling_old_weight, + sibling_reinforced_count, ) - for sibling_source, sibling_target, sibling_type, reduction in sibling_reductions_by_source.get(source_id, []) ), ) @@ -1842,10 +1995,11 @@ def record_deactivation_outcome( """, (contribution["contribution_id"],), ).fetchall() + old_weights: dict[tuple[str, str, str], float] = {} for mutation in mutation_rows: edge = connection.execute( """ - SELECT weight, reinforced_count FROM edges + SELECT weight FROM edges WHERE source_id = ? AND target_id = ? AND edge_type = ? """, ( @@ -1856,64 +2010,41 @@ def record_deactivation_outcome( ).fetchone() if edge is None: raise KeyError("journaled contribution edge is absent") - old_weight = float(edge["weight"]) - signed_delta = float(mutation["actual_delta"]) - if str(mutation["mutation_role"]) == "credited": - new_weight = max( - float(contribution["baseline_weight"]), - old_weight - signed_delta, - ) - new_count = max( - 0, - int(edge["reinforced_count"]) - - int(contribution["reinforced_count_delta"]), - ) - connection.execute( - """ - UPDATE edges SET weight = ?, reinforced_count = ? - WHERE source_id = ? AND target_id = ? AND edge_type = ? - """, - ( - new_weight, - new_count, - mutation["source_id"], - mutation["target_id"], - mutation["edge_type"], - ), - ) - else: - new_weight = old_weight - signed_delta - connection.execute( - """ - UPDATE edges SET weight = ? - WHERE source_id = ? AND target_id = ? AND edge_type = ? - """, - ( - new_weight, - mutation["source_id"], - mutation["target_id"], - mutation["edge_type"], - ), - ) + identity = ( + str(mutation["source_id"]), + str(mutation["target_id"]), + str(mutation["edge_type"]), + ) + old_weights[identity] = float(edge["weight"]) + connection.execute( + """ + UPDATE feedback_contributions + SET active = 0, reversed_by_outcome_id = ? + WHERE contribution_id = ? + """, + (outcome_id, contribution["contribution_id"]), + ) + for mutation in mutation_rows: + identity = ( + str(mutation["source_id"]), + str(mutation["target_id"]), + str(mutation["edge_type"]), + ) + old_weight = old_weights[identity] + new_weight, _ = self._rebuild_edge_from_active_journal( + connection, *identity + ) mutations.append( { "mutation_role": str(mutation["mutation_role"]), "source_id": str(mutation["source_id"]), "target_id": str(mutation["target_id"]), "edge_type": str(mutation["edge_type"]), - "actual_delta": signed_delta, + "actual_delta": float(mutation["actual_delta"]), "old_weight": old_weight, "new_weight": new_weight, } ) - connection.execute( - """ - UPDATE feedback_contributions - SET active = 0, reversed_by_outcome_id = ? - WHERE contribution_id = ? - """, - (outcome_id, contribution["contribution_id"]), - ) reversed_contributions.append( { "contribution_id": str(contribution["contribution_id"]), diff --git a/tests/test_outcome_feedback_deactivation_interleaving.py b/tests/test_outcome_feedback_deactivation_interleaving.py new file mode 100644 index 0000000..929c4db --- /dev/null +++ b/tests/test_outcome_feedback_deactivation_interleaving.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import unittest + +from neuron_graph_rag import FeedbackLedger, SourceUseEvent +from neuron_graph_rag.evidence_feedback import EngineConfig, NeuronGraphRAG + + +def _use_relation( + engine: NeuronGraphRAG, + ledger: FeedbackLedger, + *, + node_id: str, + index: int, +) -> str: + trace = engine.search_channels("alpha", limit=5, now=1_000.0 + index).relation + ledger.record_source_use( + trace.trace_id, + [ + SourceUseEvent(node_id, "selected"), + SourceUseEvent(node_id, "validated"), + SourceUseEvent(node_id, "used"), + ], + idempotency_key=f"interleaved-use-{index}", + ) + return trace.trace_id + + +class OutcomeFeedbackDeactivationInterleavingTest(unittest.TestCase): + def test_sibling_then_credited_reversal_is_bounded_and_order_reversible(self) -> None: + config = EngineConfig( + sparse_weight=1.0, + dense_weight=0.0, + seed_count=1, + max_hops=2, + feedback_learning_rate=2.0, + sibling_feedback_normalization=1.0, + maximum_edge_weight=2.0, + soft_start_feedback_reinforcement=True, + soft_start_feedback_ratio=0.25, + confirmation_decay_ratio=0.5, + outcome_driven_feedback_deactivation=True, + ) + with NeuronGraphRAG(config=config) as engine: + engine.add_document("source", "alpha lexical source") + engine.add_document("first", "first relation target") + engine.add_document("second", "second relation target") + engine.add_edge("source", "first", "supports", weight=0.5) + engine.add_edge("source", "second", "supports", weight=1.9) + ledger = FeedbackLedger(engine) + + first_trace = _use_relation( + engine, ledger, node_id="first", index=1 + ) + ledger.record_outcome( + first_trace, + ["first"], + "confirmed", + "first edge normalizes its sibling", + idempotency_key="interleaved-first-confirmed", + ) + second_trace = _use_relation( + engine, ledger, node_id="second", index=2 + ) + ledger.record_outcome( + second_trace, + ["second"], + "confirmed", + "sibling later becomes credited", + idempotency_key="interleaved-second-confirmed", + ) + + ledger.record_outcome( + first_trace, + ["first"], + "corrected", + "reverse the earlier sibling normalization", + idempotency_key="interleaved-first-corrected", + ) + second = engine.store.edge("source", "second", "supports") + self.assertAlmostEqual(second.weight, 2.0) + self.assertLessEqual(second.weight, config.maximum_edge_weight) + + ledger.record_outcome( + second_trace, + ["second"], + "corrected", + "reverse the later credited contribution", + idempotency_key="interleaved-second-corrected", + ) + self.assertAlmostEqual( + engine.store.edge("source", "second", "supports").weight, 1.9 + ) + self.assertAlmostEqual( + engine.store.edge("source", "first", "supports").weight, 0.5 + ) + + +if __name__ == "__main__": + unittest.main()