fix(kvb-org-queue): reread the config on every poll of a waiting hold

A waiting hold kept the config.json it started with, so a raised RAM or slot limit only reached it after cancel and re-hold. wait() now calls fresh() each poll; a missing, unreadable or incomplete file keeps the last complete config, so a half-saved edit never ends a wait. The History and learning details move to learning.md, bringing SKILL.md from 605 to 540 of 600 words and clearing master's red test_lock.

🎯 Fix: 92 %
💥 Break: 3 %
💸 Cost: 100 %
⏱️ Time: 100 %
🛡️ Security: 100 %
🏆 Score: 89
🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 %

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-12 11:25:24 +00:00
1 parent fd8fc08603
commit 7c0872ea68
4 files changed
+59 -16

No files matched your search

+22
View File
@@ -15,6 +15,7 @@ import time
import unittest
from collections.abc import Callable
from pathlib import Path
from unittest import mock
REPO_ROOT = Path(__file__).resolve().parents[1]
QUEUE = REPO_ROOT / "skills" / "kvb-org-queue" / "scripts" / "queue.py"
@@ -757,6 +758,27 @@ class TestQueueCommands(unittest.TestCase):
self.assertFalse(lock.exists())
self.assertEqual(self.entries(), [])
def test_a_waiting_hold_is_granted_once_the_config_it_waits_on_changes(self):
self.cfg["slot_limits"]["deploy"] = 0
self.write_config()
hold, _ = self.blocked_hold("true")
self.cfg["slot_limits"]["deploy"] = 1
self.write_config()
self.assertEqual(hold.wait(timeout=60), 0)
self.assertEqual(self.entries(), [])
def test_fresh_takes_the_new_config_and_keeps_the_last_good_one_on_a_broken_or_partial_file(self):
queue = _load(QUEUE)
good = dict(self.cfg)
new = {**self.cfg, "slot_limits": {"test": 2, "deploy": 2}}
partial = {key: value for key, value in self.cfg.items() if key != "budget_percent"}
for loaded, expected in ((new, new), (None, good), (partial, good), ([], good)):
with self.subTest(loaded=loaded), mock.patch.object(queue.watch, "load_config", return_value=loaded):
self.assertEqual(queue.fresh(good), expected)
for error in (ValueError("half-saved"), OSError("gone")):
with self.subTest(error=error), mock.patch.object(queue.watch, "load_config", side_effect=error):
self.assertEqual(queue.fresh(good), good)
def test_hold_takes_the_need_learned_for_its_command_from_both_history_files(self):
self.cfg["slot_limits"]["deploy"] = 0
self.write_config()