fix: incorporate post-merge feedback + pre-PR review fixes

DESIGN_SPEC.md

@@ -163,7 +163,7 @@ Every agent has a comprehensive identity. At the design level, agent data splits
   - `ConflictApproach`: `avoid`, `accommodate`, `compete`, `compromise`, `collaborate` (Thomas-Kilmann model)
 
 ```yaml
-# --- Current (M2): Config layer — AgentIdentity (frozen) ---
+# --- Current (M3): Config layer — AgentIdentity (frozen) ---
 agent:
   id: "uuid"
   name: "Sarah Chen"
@@ -2374,7 +2374,6 @@ ai-company/
 │       │   ├── metrics.py          # TaskCompletionMetrics proxy overhead model
 │       │   ├── react_loop.py       # ReAct loop implementation
 │       │   ├── plan_models.py      # Plan step, plan, and plan-execute config models
-│       │   ├── plan_parsing.py     # Plan response parsing utilities
 │       │   ├── plan_execute_loop.py # Plan-and-Execute loop implementation
 │       │   ├── plan_parsing.py     # Plan extraction from LLM responses (JSON + text fallback)
 │       │   ├── loop_helpers.py     # Shared stateless helpers for all loop implementations
@@ -2386,7 +2385,7 @@ ai-company/
 │       │   ├── parallel_models.py  # AgentAssignment, ParallelExecutionGroup, AgentOutcome, ParallelExecutionResult, ParallelProgress
 │       │   ├── resource_lock.py    # ResourceLock protocol + InMemoryResourceLock
 │       │   ├── shutdown.py        # Graceful shutdown strategy & manager
-│       │   ├── task_engine.py      # Task routing & scheduling (M3-M4)
+│       │   ├── task_engine.py      # Task routing & scheduling (M4)
 │       │   ├── workflow_engine.py  # Workflow orchestration (M4)
 │       │   ├── meeting_engine.py   # Meeting coordination (M4)
 │       │   └── hr_engine.py        # Hiring, firing, performance (M7)
@@ -2482,6 +2481,7 @@ ai-company/
 │       │   ├── permissions.py      # ToolPermissionChecker (access-level gating)
 │       │   ├── errors.py           # Tool error hierarchy (incl. ToolPermissionDeniedError)
 │       │   ├── examples/           # Example tool implementations
+│       │   │   ├── __init__.py    # Package exports
 │       │   │   └── echo.py        # Echo tool (for testing)
 │       │   ├── sandbox/            # Sandboxing backends
 │       │   │   ├── __init__.py    # Package exports
@@ -2502,8 +2502,8 @@ ai-company/
 │       │   ├── _git_base.py        # Base class for git tools (workspace, subprocess, sandbox integration)
 │       │   ├── _process_cleanup.py  # Subprocess transport cleanup utility (Windows ResourceWarning prevention)
 │       │   ├── git_tools.py        # Git operations — 6 built-in tools (sandbox-aware)
-│       │   ├── code_runner.py      # Code execution (M3)
-│       │   ├── web_tools.py        # HTTP, search (M3)
+│       │   ├── code_runner.py      # Code execution (M7)
+│       │   ├── web_tools.py        # HTTP, search (M7)
 │       │   └── mcp_bridge.py       # MCP server integration (M7)
 │       ├── security/                # Security & approval (M7, stubs only)
 │       │   ├── approval.py         # Approval workflow (M7)
@@ -2597,7 +2597,7 @@ These conventions were established during the M0–M2+ review cycle. **Adopted**
 | **State coordination** | Planned (M4) | Centralized single-writer: `TaskEngine` owns all task/project mutations via `asyncio.Queue`. Agents submit requests, engine applies `model_copy(update=...)` sequentially and publishes snapshots. `version: int` field on state models for future optimistic concurrency if multi-process scaling is needed. | Prevents lost updates by design. Trivial in single-threaded asyncio (no locks). Perfect audit trail. Industry consensus: MetaGPT, CrewAI, AutoGen all use prevention-by-design, not conflict resolution. See §6.8 State Coordination table. |
 | **Workspace isolation** | Planned (M4) | Pluggable `WorkspaceIsolationStrategy` protocol. Default: planner + git worktrees. Each agent works in an isolated worktree; sequential merge on completion. Textual conflicts detected by git; semantic conflicts reviewed by agent or human. | Industry standard (Codex, Cursor, Claude Code, VS Code). Maximum parallelism. Leverages mature git infrastructure. See §6.8. |
 | **Graceful shutdown** | Adopted (M3) | Pluggable `ShutdownStrategy` protocol. Default: cooperative with 30s timeout. Agents check shutdown event at turn boundaries. Force-cancel after timeout. `INTERRUPTED` status for force-cancelled tasks. M4/M5: upgrade to checkpoint-and-stop. | Cross-platform (Windows `signal.signal()` fallback). Bounded shutdown time. Mirrors cooperative shutdown in §6.7. |
-| **Template inheritance** | Adopted (M2.5) | `extends` field on `CompanyTemplate` triggers parent resolution at render time. `merge.py` merges configs by field type: scalars (child wins), config dicts (deep merge), agents (by `(role, department)` key with `_remove` support), departments (by name). `_ParentEntry` dataclass tracks merge state. `DEFAULT_MERGE_DEPARTMENT = "engineering"` shared between merge and renderer. Circular chains detected via `frozenset` tracking; max depth = 10. | Enables template composition without copy-paste. Merge-by-key preserves parent order. `_remove` directive enables clean agent removal without workarounds. |
+| **Template inheritance** | Adopted (M2.5) | `extends` field on `CompanyTemplate` triggers parent resolution at render time. `merge.py` merges configs by field type: scalars (child wins), config dicts (deep merge), agents (by `(role, department, merge_id)` key with `_remove` support), departments (by name). `_ParentEntry` dataclass tracks merge state. `DEFAULT_MERGE_DEPARTMENT = "engineering"` shared between merge and renderer. Circular chains detected via `frozenset` tracking; max depth = 10. | Enables template composition without copy-paste. Merge-by-key preserves parent order. `_remove` directive enables clean agent removal without workarounds. |
 | **Pydantic alias for YAML directives** | Adopted (M2.5) | `Field(alias="_remove")` in `TemplateAgentConfig` — YAML uses `_remove: true`, Python accesses `agent.remove`. Keeps the YAML-facing name (underscore prefix signals internal directive) separate from the Python attribute name. | Underscore-prefixed YAML keys signal merge directives vs regular fields. Pydantic alias bridges the naming convention gap cleanly. |
 | **Communication foundation** | Adopted (M4) | `MessageBus` protocol with `InMemoryMessageBus` backend (asyncio queues, pull-model `receive()` with shutdown signaling via `asyncio.Event`). `MessageDispatcher` routes to concurrent handlers via `asyncio.TaskGroup` with pre-allocated error collection. `AgentMessenger` per-agent facade auto-fills sender/timestamp/ID; deterministic direct-channel naming `@{sorted_a}:{sorted_b}`. `DeliveryEnvelope` for delivery tracking. `NotBlankStr` validation on all protocol boundary identifiers. | Pull-model avoids callback complexity and enables agents to consume at their own pace. Protocol + backend split enables future persistent/distributed bus implementations. Deterministic DM channel names prevent duplicates. See §5. |
 | **Delegation & loop prevention** | Adopted (M4) | `HierarchyResolver` resolves org hierarchy from `Company` at construction (cycle-detected, `MappingProxyType`-frozen). `AuthorityValidator` checks chain-of-command + role permissions. `DelegationGuard` orchestrates five mechanisms (ancestry, depth, dedup, rate limit, circuit breaker) in sequence, short-circuiting on first rejection. `DelegationService` is synchronous (CPU-only); messaging integration deferred. Stateful mechanisms use injectable clock for deterministic testing. Task model extended with `parent_task_id` and `delegation_chain` fields. | Synchronous delegation avoids async complexity for CPU-only validation. Five-mechanism guard provides defence-in-depth against all loop patterns. Injectable clocks enable deterministic testing. See §5.4, §5.5. |

src/ai_company/communication/bus_memory.py

@@ -5,6 +5,7 @@
 """
 
 import asyncio
+import contextlib
 from collections import deque
 from datetime import UTC, datetime
 from typing import NoReturn
@@ -38,11 +39,14 @@
     COMM_HISTORY_QUERIED,
     COMM_MESSAGE_DELIVERED,
     COMM_MESSAGE_PUBLISHED,
+    COMM_RECEIVE_SHUTDOWN,
     COMM_RECEIVE_TIMEOUT,
+    COMM_RECEIVE_UNSUBSCRIBED,
     COMM_SEND_DIRECT_INVALID,
     COMM_SUBSCRIPTION_CREATED,
     COMM_SUBSCRIPTION_NOT_FOUND,
     COMM_SUBSCRIPTION_REMOVED,
+    COMM_UNSUBSCRIBE_SENTINEL_FAILED,
 )
 
 logger = get_logger(__name__)
@@ -93,7 +97,7 @@ def __init__(self, *, config: MessageBusConfig) -> None:
         self._config = config
         self._lock = asyncio.Lock()
         self._channels: dict[str, Channel] = {}
-        self._queues: dict[tuple[str, str], asyncio.Queue[DeliveryEnvelope]] = {}
+        self._queues: dict[tuple[str, str], asyncio.Queue[DeliveryEnvelope | None]] = {}
         self._history: dict[str, deque[Message]] = {}
         self._known_agents: set[str] = set()
         self._running = False
@@ -154,7 +158,7 @@ def _ensure_queue(
         self,
         channel_name: str,
         subscriber_id: str,
-    ) -> asyncio.Queue[DeliveryEnvelope]:
+    ) -> asyncio.Queue[DeliveryEnvelope | None]:
         """Get or create a per-(channel, subscriber) queue."""
         return self._queues.setdefault(
             (channel_name, subscriber_id),
@@ -395,7 +399,17 @@ async def unsubscribe(
             self._channels[channel_name] = channel.model_copy(
                 update={"subscribers": new_subs},
             )
-            self._queues.pop((channel_name, subscriber_id), None)
+            queue = self._queues.pop((channel_name, subscriber_id), None)
+            if queue is not None:
+                try:
+                    queue.put_nowait(None)
+                except asyncio.QueueFull:
+                    logger.exception(
+                        COMM_UNSUBSCRIBE_SENTINEL_FAILED,
+                        channel=channel_name,
+                        subscriber=subscriber_id,
+                        detail="Queue full — unsubscribe sentinel not delivered",
+                    )
         logger.info(
             COMM_SUBSCRIPTION_REMOVED,
             channel=channel_name,
@@ -415,13 +429,19 @@ async def receive(
         the bus is stopped.  When ``timeout`` is ``None``, awaits
         indefinitely (or until shutdown).
 
+        Note: Only one ``receive()`` call should be pending per
+        ``(channel_name, subscriber_id)`` pair at a time.  The
+        unsubscribe sentinel wakes a single waiter; concurrent
+        receivers on the same subscription are not supported.
+
         Args:
             channel_name: Channel to receive from.
             subscriber_id: Agent ID receiving.
             timeout: Seconds to wait before returning ``None``.
 
         Returns:
-            A delivery envelope, or ``None`` on timeout or shutdown.
+            A delivery envelope, or ``None`` on timeout, shutdown,
+            or when an in-flight receive is woken by :meth:`unsubscribe`.
 
         Raises:
             MessageBusNotRunningError: If the bus is not running.
@@ -442,17 +462,39 @@ async def receive(
             queue = self._ensure_queue(channel_name, subscriber_id)
         result = await self._await_with_shutdown(queue, timeout)
         if result is None:
+            self._log_receive_null(channel_name, subscriber_id, timeout)
+        return result
+
+    def _log_receive_null(
+        self,
+        channel_name: str,
+        subscriber_id: str,
+        timeout: float | None,
+    ) -> None:
+        """Log the cause when ``receive()`` returns ``None``."""
+        if self._shutdown_event.is_set():
+            logger.debug(
+                COMM_RECEIVE_SHUTDOWN,
+                channel=channel_name,
+                subscriber=subscriber_id,
+            )
+        elif (channel_name, subscriber_id) not in self._queues:
+            logger.debug(
+                COMM_RECEIVE_UNSUBSCRIBED,
+                channel=channel_name,
+                subscriber=subscriber_id,
+            )
+        else:
             logger.debug(
                 COMM_RECEIVE_TIMEOUT,
                 channel=channel_name,
                 subscriber=subscriber_id,
                 timeout=timeout,
             )
-        return result
 
     async def _await_with_shutdown(
         self,
-        queue: asyncio.Queue[DeliveryEnvelope],
+        queue: asyncio.Queue[DeliveryEnvelope | None],
         timeout: float | None,  # noqa: ASYNC109
     ) -> DeliveryEnvelope | None:
         """Await next envelope, returning ``None`` on timeout or shutdown.
@@ -464,8 +506,8 @@ async def _await_with_shutdown(
         Returns:
             The next envelope, or ``None``.
         """
-        get_task = asyncio.ensure_future(queue.get())
-        shutdown_task = asyncio.ensure_future(
+        get_task = asyncio.create_task(queue.get())
+        shutdown_task = asyncio.create_task(
             self._shutdown_event.wait(),
         )
         try:
@@ -480,8 +522,12 @@ async def _await_with_shutdown(
             raise
         if not get_task.done():
             get_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await get_task
         if not shutdown_task.done():
             shutdown_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await shutdown_task
         if get_task in done and not get_task.cancelled():
             return get_task.result()
         return None

src/ai_company/communication/dispatcher.py

@@ -4,6 +4,7 @@
 """
 
 import asyncio
+import inspect
 from uuid import UUID  # noqa: TC003 -- required at runtime by Pydantic
 
 from pydantic import BaseModel, ConfigDict, Field, computed_field
@@ -28,6 +29,7 @@
     COMM_DISPATCH_START,
     COMM_HANDLER_DEREGISTER_MISS,
     COMM_HANDLER_DEREGISTERED,
+    COMM_HANDLER_INVALID,
     COMM_HANDLER_REGISTERED,
 )
 
@@ -96,6 +98,19 @@ def register(
         """
         if not isinstance(handler, MessageHandler):
             handler = FunctionHandler(handler)
+        elif not inspect.iscoroutinefunction(handler.handle):
+            msg = (
+                f"MessageHandler {type(handler).__name__!r} has a "
+                f"synchronous handle() — must be async"
+            )
+            logger.warning(
+                COMM_HANDLER_INVALID,
+                agent_id=self._agent_id,
+                handler_name=name,
+                handler_type=type(handler).__name__,
+                error=msg,
+            )
+            raise TypeError(msg)
 
         registration = HandlerRegistration(
             handler=handler,

src/ai_company/communication/messenger.py

@@ -10,7 +10,10 @@
     MessageHandlerFunc,
 )
 from ai_company.communication.message import Message
-from ai_company.communication.subscription import Subscription  # noqa: TC001
+from ai_company.communication.subscription import (  # noqa: TC001
+    DeliveryEnvelope,
+    Subscription,
+)
 from ai_company.observability import get_logger
 from ai_company.observability.events.communication import (
     COMM_DISPATCH_NO_DISPATCHER,
@@ -269,6 +272,28 @@ async def unsubscribe(self, channel_name: str) -> None:
             channel=channel_name,
         )
 
+    async def receive(
+        self,
+        channel_name: str,
+        *,
+        timeout: float | None = None,  # noqa: ASYNC109
+    ) -> DeliveryEnvelope | None:
+        """Receive the next message from a channel.
+
+        Args:
+            channel_name: Channel to receive from.
+            timeout: Max seconds to wait, or ``None`` for indefinite.
+
+        Returns:
+            The next delivery envelope, or ``None`` on timeout, shutdown,
+            or when an in-flight receive is woken by :meth:`unsubscribe`.
+        """
+        return await self._bus.receive(
+            channel_name,
+            self._agent_id,
+            timeout=timeout,
+        )
+
     def register_handler(
         self,
         handler: MessageHandler | MessageHandlerFunc,

src/ai_company/core/company.py

@@ -38,6 +38,7 @@ def _validate_not_self_report(self) -> Self:
                 f"Agent cannot report to themselves: "
                 f"{self.subordinate!r} == {self.supervisor!r}"
             )
+            logger.warning(COMPANY_VALIDATION_ERROR, error=msg)
             raise ValueError(msg)
         return self
 
@@ -242,10 +243,13 @@ class Team(BaseModel):
 
     @model_validator(mode="after")
     def _validate_no_duplicate_members(self) -> Self:
-        """Ensure no duplicate members."""
-        if len(self.members) != len(set(self.members)):
-            dupes = sorted(m for m, c in Counter(self.members).items() if c > 1)
+        """Ensure no duplicate members (case-insensitive)."""
+        normalized = [m.strip().casefold() for m in self.members]
+        if len(normalized) != len(set(normalized)):
+            dup_keys = {m for m, c in Counter(normalized).items() if c > 1}
+            dupes = sorted(m for m in self.members if m.strip().casefold() in dup_keys)
             msg = f"Duplicate members in team {self.name!r}: {dupes}"
+            logger.warning(COMPANY_VALIDATION_ERROR, error=msg)
             raise ValueError(msg)
         return self
 
@@ -291,11 +295,15 @@ class Department(BaseModel):
 
     @model_validator(mode="after")
     def _validate_unique_team_names(self) -> Self:
-        """Ensure no duplicate team names within a department."""
-        names = [t.name for t in self.teams]
+        """Ensure no duplicate team names within a department (case-insensitive)."""
+        names = [t.name.strip().casefold() for t in self.teams]
         if len(names) != len(set(names)):
-            dupes = sorted(n for n, c in Counter(names).items() if c > 1)
+            dup_keys = {n for n, c in Counter(names).items() if c > 1}
+            dupes = sorted(
+                t.name for t in self.teams if t.name.strip().casefold() in dup_keys
+            )
             msg = f"Duplicate team names in department {self.name!r}: {dupes}"
+            logger.warning(COMPANY_VALIDATION_ERROR, error=msg)
             raise ValueError(msg)
         return self
 
@@ -309,6 +317,7 @@ def _validate_unique_subordinates(self) -> Self:
                 f"Duplicate subordinates in reporting lines "
                 f"for department {self.name!r}: {dupes}"
             )
+            logger.warning(COMPANY_VALIDATION_ERROR, error=msg)
             raise ValueError(msg)
         return self
 
@@ -375,11 +384,15 @@ class HRRegistry(BaseModel):
 
     @model_validator(mode="after")
     def _validate_no_duplicate_active_agents(self) -> Self:
-        """Ensure no duplicate entries in active_agents."""
-        agents = self.active_agents
-        if len(agents) != len(set(agents)):
-            dupes = sorted(a for a, c in Counter(agents).items() if c > 1)
+        """Ensure no duplicate entries in active_agents (case-insensitive)."""
+        normalized = [a.strip().casefold() for a in self.active_agents]
+        if len(normalized) != len(set(normalized)):
+            dup_keys = {a for a, c in Counter(normalized).items() if c > 1}
+            dupes = sorted(
+                a for a in self.active_agents if a.strip().casefold() in dup_keys
+            )
             msg = f"Duplicate entries in active_agents: {dupes}"
+            logger.warning(COMPANY_VALIDATION_ERROR, error=msg)
             raise ValueError(msg)
         return self