fix: tool registry emits Anthropic schema, question lifecycle cleanup

- Tool.to_schema() emits input_schema (Anthropic Messages API) instead
  of OpenAI-style parameters
- QuestionManager: remove questions from _pending on answer and timeout,
  reject double answers, add list_pending()
- core/skills/__init__.py so skills are importable as a package
- revert optional-import fallbacks in logger/message_bus: nats-py and
  loguru are hard requirements, a broken install should fail loudly
- tests for wait_for_answer happy path and timeout cleanup

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012AJLYcbTHCe45fyFnMfEin

.env.example

@@ -0,0 +1,16 @@
+GO_API_BASE_URL=
+NATS_URL=
+REDIS_HOST=
+REDIS_PORT=
+REDIS_PASSWORD=
+INTERNAL_API_KEY=
+DB_HOST=
+DB_PORT=
+DB_NAME=
+DB_USER=
+DB_PASSWORD=
+NATS_USER=
+NATS_PASSWORD=
+ANTHROPIC_API_KEY=
+LLM_MODEL=claude-opus-4-8
+LOG_LEVEL=INFO

.gitignore

@@ -7,3 +7,8 @@ venv/
 
 .idea/
 .vscode/
+
+# secrets
+.env
+.env.*
+!.env.example

config/system_config.py

@@ -1,27 +1,28 @@
 """System configuration for LogiFlow AI."""
 import os
 
-# Infrastructure Connections — values come from environment / .env file.
-# Hardcoded strings are last-resort defaults; set the matching env var in production.
-NATS_URL      = os.getenv("NATS_URL",      "nats://doormile:Package@321#@66.116.226.161:4223")
-NATS_HOST     = os.getenv("NATS_HOST",     "66.116.226.161")
-NATS_PORT     = int(os.getenv("NATS_PORT", "4223"))
-NATS_USER     = os.getenv("NATS_USER",     "doormile")
-NATS_PASSWORD = os.getenv("NATS_PASSWORD", "Package@321#")
+# Infrastructure Connections — values come from environment / .env file
+# (see .env.example). Defaults point at localhost with no credentials so the
+# system runs in local-fallback mode; real hosts and secrets are never in code.
+NATS_URL      = os.getenv("NATS_URL",      "nats://localhost:4222")
+NATS_HOST     = os.getenv("NATS_HOST",     "localhost")
+NATS_PORT     = int(os.getenv("NATS_PORT", "4222"))
+NATS_USER     = os.getenv("NATS_USER",     "")
+NATS_PASSWORD = os.getenv("NATS_PASSWORD", "")
 
-REDIS_HOST     = os.getenv("REDIS_HOST",     "66.116.226.255")
-REDIS_PORT     = int(os.getenv("REDIS_PORT", "6380"))
-REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "Package@321#")
+REDIS_HOST     = os.getenv("REDIS_HOST",     "localhost")
+REDIS_PORT     = int(os.getenv("REDIS_PORT", "6379"))
+REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
 
 # Postgres — individual params to avoid @ in password breaking DSN parsing
-DB_HOST     = os.getenv("DB_HOST",     "31.97.228.132")
-DB_PORT     = int(os.getenv("DB_PORT", "5433"))
+DB_HOST     = os.getenv("DB_HOST",     "localhost")
+DB_PORT     = int(os.getenv("DB_PORT", "5432"))
 DB_NAME     = os.getenv("DB_NAME",     "logistics")
-DB_USER     = os.getenv("DB_USER",     "admin")
-DB_PASSWORD = os.getenv("DB_PASSWORD", "Package@321#")
+DB_USER     = os.getenv("DB_USER",     "postgres")
+DB_PASSWORD = os.getenv("DB_PASSWORD", "")
 
 GO_API_BASE_URL  = os.getenv("GO_API_BASE_URL",  "http://localhost:8080")
-INTERNAL_API_KEY = os.getenv("INTERNAL_API_KEY", "doormile-internal-2024")
+INTERNAL_API_KEY = os.getenv("INTERNAL_API_KEY", "")
 
 # Route Optimization API (Valhalla-backed road routing). The ExpressDispatchAgent
 # calls its Doormile endpoint (/api/v1/optimization/doormile/sequence) to order a

core/interactive_question.py

@@ -110,24 +110,33 @@ class QuestionManager:
             raise TimeoutError(f"Question {question.question_id} timed out waiting for human input")
         finally:
             self._futures.pop(question.question_id, None)
+            self._pending.pop(question.question_id, None)
 
     def answer_question(self, question_id: str, answer: Any) -> bool:
-        """Called when human operator submits an answer from the UI."""
+        """Called when human operator submits an answer from the UI.
+
+        Returns False if the question is unknown or was already answered."""
         question = self._pending.get(question_id)
         if not question:
             logger.warning(f"No pending question found for id {question_id}")
             return False
+        if question.answered:
+            logger.warning(f"Question {question_id} already answered; ignoring")
+            return False
 
         question.answered = True
         question.answer = answer
 
         fut = self._futures.get(question_id)
         if fut and not fut.done():
-            fut.set_result(answer)
-            return True
-
+            fut.set_result(answer)  # wait_for_answer() removes it from _pending
+        else:
+            self._pending.pop(question_id, None)  # nobody waiting; don't leak
         return True
 
+    def list_pending(self) -> List[Question]:
+        return [q for q in self._pending.values() if not q.answered]
+
     def get_pending(self, question_id: str) -> Optional[Question]:
         return self._pending.get(question_id)
 

core/logger.py

@@ -1,33 +1,27 @@
-"""Structured logging for LogiFlow AI — loguru setup with standard logging fallback."""
+"""Structured logging for LogiFlow AI — single loguru setup imported by all modules."""
 import sys
 import os
-import logging
 from pathlib import Path
+from loguru import logger
 
-try:
-    from loguru import logger
+Path("logs").mkdir(exist_ok=True)
 
-    Path("logs").mkdir(exist_ok=True)
+logger.remove()
 
-    logger.remove()
+logger.add(
+    sys.stderr,
+    format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
+    level=os.getenv("LOG_LEVEL", "INFO"),
+    colorize=False,
+)
 
-    logger.add(
-        sys.stderr,
-        format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
-        level=os.getenv("LOG_LEVEL", "INFO"),
-        colorize=False,
-    )
-
-    logger.add(
-        "logs/logiflow_{time:YYYY-MM-DD}.log",
-        rotation="100 MB",
-        retention="30 days",
-        compression="gz",
-        level="DEBUG",
-        format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
-    )
-except ImportError:
-    logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)-8s | %(message)s")
-    logger = logging.getLogger("logiflow")
+logger.add(
+    "logs/logiflow_{time:YYYY-MM-DD}.log",
+    rotation="100 MB",
+    retention="30 days",
+    compression="gz",
+    level="DEBUG",
+    format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
+)
 
 __all__ = ["logger"]

core/message_bus.py

@@ -8,11 +8,8 @@ from collections import defaultdict
 from dataclasses import dataclass
 from enum import Enum
 
-try:
-    import nats
-    import nats.js.errors
-except ImportError:
-    nats = None
+import nats
+import nats.js.errors
 
 from core.types import AgentMessage, MessageType
 from core.logger import logger
@@ -59,9 +56,6 @@ class MessageBus:
 
     async def connect(self):
         """Connect to NATS and create the logistics JetStream stream."""
-        if nats is None:
-            logger.warning("nats-py not installed; message bus running in local in-memory mode")
-            return
         self._nc = await nats.connect(
             servers=[f"nats://{NATS_HOST}:{NATS_PORT}"],
             user=NATS_USER,

core/tool_registry.py

@@ -33,7 +33,7 @@ class Tool:
     metadata: Dict[str, Any] = field(default_factory=dict)
 
     def to_schema(self) -> Dict[str, Any]:
-        """Generate OpenAI/Claude compatible JSON Schema for tool calling."""
+        """Generate an Anthropic Messages API tool definition (name / description / input_schema)."""
         properties = {}
         required = []
 
@@ -56,7 +56,7 @@ class Tool:
         return {
             "name": self.name,
             "description": self.description,
-            "parameters": {
+            "input_schema": {
                 "type": "object",
                 "properties": properties,
                 "required": required,

doormile_test.py

@@ -3,6 +3,7 @@
 Doormile Full System Test Suite v2
 Fixed: correct routes, 2-step login, pricing payload, health endpoint
 """
+import os
 import asyncio
 import aiohttp
 import json
@@ -13,14 +14,17 @@ import redis.asyncio as aioredis
 from datetime import datetime
 
 # ── Config ────────────────────────────────────────────────────────────────────
-API_BASE       = "https://api.doormile.com"
-NATS_URL       = "nats://66.116.226.161:4223"
-NATS_USER      = "doormile"
-NATS_PASSWORD  = "Package@321#"
-REDIS_HOST     = "66.116.226.255"
-REDIS_PORT     = 6380
-REDIS_PASSWORD = "Package@321#"
-INTERNAL_KEY   = "doormile-internal-2024"
+from dotenv import load_dotenv
+load_dotenv()
+
+API_BASE       = os.getenv("GO_API_BASE_URL", "https://api.doormile.com")
+NATS_URL       = os.environ["NATS_URL"]
+NATS_USER      = os.getenv("NATS_USER", "")
+NATS_PASSWORD  = os.getenv("NATS_PASSWORD", "")
+REDIS_HOST     = os.environ["REDIS_HOST"]
+REDIS_PORT     = int(os.getenv("REDIS_PORT", "6379"))
+REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
+INTERNAL_KEY   = os.environ["INTERNAL_API_KEY"]
 
 # Test customer — uses the one created by Window 1 test
 TEST_PHONE     = "9900000001"

main.py

@@ -204,11 +204,12 @@ Usage:
     python main.py --portal         Launch Streamlit customer portal
     python main.py --help           Show this help
 
-Infrastructure (set in .env):
-    NATS   nats://doormile@66.116.226.161:4223
-    Redis  66.116.226.255:6380
+Infrastructure (set in .env — see .env.example):
+    NATS   NATS_URL / NATS_USER / NATS_PASSWORD
+    Redis  REDIS_HOST / REDIS_PORT / REDIS_PASSWORD
     PG     DB_HOST / DB_PORT / DB_NAME / DB_USER / DB_PASSWORD
-    API    GO_API_BASE_URL
+    API    GO_API_BASE_URL / INTERNAL_API_KEY
+    LLM    ANTHROPIC_API_KEY / LLM_MODEL
     """)
 
 

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+asyncio_mode = auto

tests/test_tool_registry.py

@@ -27,7 +27,8 @@ async def test_tool_registry_registration_and_execution():
     schemas = registry.get_schemas()
     assert len(schemas) == 1
     assert schemas[0]["name"] == "test_tool"
-    assert "order_id" in schemas[0]["parameters"]["properties"]
+    assert "order_id" in schemas[0]["input_schema"]["properties"]
+    assert "parameters" not in schemas[0]  # Anthropic shape, not OpenAI
 
     result = await registry.execute_tool("test_tool", order_id="ORD-100")
     assert result == {"order": "ORD-100", "items": 1}
@@ -50,6 +51,32 @@ async def test_question_manager_flow():
     assert q.options[0].badge == "12 AM–9 AM"
     assert not q.answered
 
-    qm.answer_question(q.question_id, "morning")
+    assert qm.list_pending() == [q]
+    assert qm.answer_question(q.question_id, "morning") is True
     assert q.answered is True
     assert q.answer == "morning"
+    assert qm.get_pending(q.question_id) is None  # cleaned up, not leaked
+    assert qm.answer_question(q.question_id, "afternoon") is False  # double-answer rejected
+
+
+@pytest.mark.asyncio
+async def test_question_manager_wait_and_answer():
+    qm = QuestionManager()
+    q = qm.create_question(prompt="Reassign?", options=["yes", "no"], allow_custom_input=False)
+
+    async def operator():
+        await asyncio.sleep(0.01)
+        qm.answer_question(q.question_id, "yes")
+
+    asyncio.create_task(operator())
+    assert await qm.wait_for_answer(q, timeout_s=1.0) == "yes"
+    assert qm.get_pending(q.question_id) is None
+
+
+@pytest.mark.asyncio
+async def test_question_manager_timeout_cleans_up():
+    qm = QuestionManager()
+    q = qm.create_question(prompt="Reassign?", options=["yes", "no"])
+    with pytest.raises(TimeoutError):
+        await qm.wait_for_answer(q, timeout_s=0.01)
+    assert qm.get_pending(q.question_id) is None