fix(linear): address PR #87 must-fix review items

- linear_reactions: guard auth-circuit globals with `_auth_state_lock`
  so the daemon sweep thread and the main thread can't race the
  read-modify-write on `_consecutive_auth_failures` /
  `_auth_circuit_open`.
- linear_reactions: wrap the daemon sweep target in
  `_sweep_stale_reactions_safe` so an unexpected exception logs at
  ERROR instead of dying silently (stderr from a daemon thread doesn't
  reliably reach CloudWatch).
- linear_reactions: only increment the sweep delete counter when
  `_graphql(_DELETE_MUTATION, ...)` actually returns a non-None
  response — previously the summary log overstated success.
- config: hoist `import boto3` out of the catch-narrowed try/except
  so an `ImportError` (boto3 missing from the image) degrades to a
  WARN log instead of crashing the agent.
- orchestrate-task: wrap `notifyLinearOnConcurrencyCap` in a
  defensive try/catch — durable-execution retries the entire
  admission-control step on throw, which would re-fire `failTask` +
  `emitTaskEvent` and produce duplicate events.
- tests: 1 new throw-propagation test for `notifyLinearOnConcurrencyCap`,
  3 new tests for `resolve_linear_api_token` (cached env, no-arn,
  ImportError fallback). Auto-reset fixture in
  `test_linear_reactions.py` now also resets the circuit-breaker
  globals between tests so future cases don't leak state.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bgagent

agent/src/config.py

@@ -60,7 +60,14 @@ def resolve_linear_api_token() -> str:
     try:
         import boto3
         from botocore.exceptions import BotoCoreError, ClientError
+    except ImportError as e:
+        # boto3 missing from the container image — degrade gracefully rather
+        # than hard-crashing the agent. The Linear MCP will fail on first
+        # call with a clear auth error.
+        log("WARN", f"resolve_linear_api_token: boto3 unavailable ({e}); skipping")
+        return ""
 
+    try:
         region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
         client = boto3.client("secretsmanager", region_name=region)
         resp = client.get_secret_value(SecretId=secret_arn)

agent/src/linear_reactions.py

@@ -95,10 +95,12 @@
 #: Linear's quota. After ``_AUTH_FAILURE_THRESHOLD`` consecutive 401/403
 #: responses, ``_auth_circuit_open`` flips to True and all later calls
 #: short-circuit (return None) without hitting the network. A successful
-#: 2xx response resets the counter.
+#: 2xx response resets the counter. The lock guards the read-modify-write
+#: against the daemon sweep thread.
 _AUTH_FAILURE_THRESHOLD = 3
 _consecutive_auth_failures = 0
 _auth_circuit_open = False
+_auth_state_lock = threading.Lock()
 
 
 def _enabled(channel_source: str, channel_metadata: dict[str, str] | None) -> str | None:
@@ -126,8 +128,9 @@ def _graphql(query: str, variables: dict[str, Any]) -> dict[str, Any] | None:
     """
     global _consecutive_auth_failures, _auth_circuit_open
 
-    if _auth_circuit_open:
-        return None
+    with _auth_state_lock:
+        if _auth_circuit_open:
+            return None
 
     token = os.environ.get("LINEAR_API_TOKEN", "")
     if not token:
@@ -149,13 +152,19 @@ def _graphql(query: str, variables: dict[str, Any]) -> dict[str, Any] | None:
         return None
 
     if resp.status_code in (401, 403):
-        _consecutive_auth_failures += 1
-        if _consecutive_auth_failures >= _AUTH_FAILURE_THRESHOLD and not _auth_circuit_open:
-            _auth_circuit_open = True
+        with _auth_state_lock:
+            _consecutive_auth_failures += 1
+            opened = (
+                _consecutive_auth_failures >= _AUTH_FAILURE_THRESHOLD and not _auth_circuit_open
+            )
+            if opened:
+                _auth_circuit_open = True
+                failures = _consecutive_auth_failures
+        if opened:
             log(
                 "ERROR",
                 "linear_reactions: auth circuit OPEN after "
-                f"{_consecutive_auth_failures} consecutive {resp.status_code}s — "
+                f"{failures} consecutive {resp.status_code}s — "
                 "API token likely revoked. Suppressing further Linear calls "
                 "for this container.",
             )
@@ -169,7 +178,8 @@ def _graphql(query: str, variables: dict[str, Any]) -> dict[str, Any] | None:
 
     # Successful 2xx — reset the auth failure counter so transient blips don't
     # accumulate toward the threshold.
-    _consecutive_auth_failures = 0
+    with _auth_state_lock:
+        _consecutive_auth_failures = 0
 
     body = resp.json() if resp.content else {}
     if body.get("errors"):
@@ -199,6 +209,23 @@ def _get_viewer_id() -> str | None:
     return None
 
 
+def _sweep_stale_reactions_safe(issue_id: str, exclude_id: str | None = None) -> None:
+    """Top-level wrapper for the sweep daemon thread.
+
+    Catches everything so an unexpected ``TypeError`` / ``AttributeError``
+    inside ``_sweep_stale_reactions`` doesn't kill the thread silently —
+    stderr from a daemon thread may not reach CloudWatch in containerized
+    environments.
+    """
+    try:
+        _sweep_stale_reactions(issue_id, exclude_id=exclude_id)
+    except Exception as e:
+        log(
+            "ERROR",
+            f"linear_reactions: sweep thread crashed ({type(e).__name__}): {e}",
+        )
+
+
 def _sweep_stale_reactions(issue_id: str, exclude_id: str | None = None) -> None:
     """Delete bgagent-owned 👀/✅/❌ reactions on the issue.
 
@@ -252,8 +279,8 @@ def _sweep_stale_reactions(issue_id: str, exclude_id: str | None = None) -> None
         if exclude_id is not None and rid == exclude_id:
             # The 👀 we just posted — skip, it's the new marker.
             continue
-        _graphql(_DELETE_MUTATION, {"id": rid})
-        deletes += 1
+        if _graphql(_DELETE_MUTATION, {"id": rid}) is not None:
+            deletes += 1
     deletes_ms = int((time.monotonic() - deletes_start) * 1000)
     total_ms = int((time.monotonic() - sweep_start) * 1000)
     log(
@@ -311,7 +338,7 @@ def react_task_started(
     # status. The sweep filters out the just-posted reaction id so it
     # never deletes itself.
     threading.Thread(
-        target=_sweep_stale_reactions,
+        target=_sweep_stale_reactions_safe,
         args=(issue_id,),
         kwargs={"exclude_id": rid},
         daemon=True,

agent/tests/test_config.py

@@ -1,8 +1,11 @@
 """Unit tests for config.py — build_config and constants."""
 
+import sys
+from unittest.mock import patch
+
 import pytest
 
-from config import PR_TASK_TYPES, build_config
+from config import PR_TASK_TYPES, build_config, resolve_linear_api_token
 from models import TaskConfig
 
 
@@ -85,3 +88,35 @@ def test_auto_generated_task_id(self):
         )
         assert config.task_id
         assert len(config.task_id) == 12
+
+
+class TestResolveLinearApiToken:
+    """Coverage for the secrets-manager + boto3 fallback paths."""
+
+    def test_returns_cached_env_var_without_calling_boto(self, monkeypatch):
+        monkeypatch.setenv("LINEAR_API_TOKEN", "lin_cached")
+        monkeypatch.setenv("LINEAR_API_TOKEN_SECRET_ARN", "arn:aws:sm:::secret/linear")
+        # boto3 must not be touched if the env var is already set.
+        with patch("config.log") as mock_log:
+            assert resolve_linear_api_token() == "lin_cached"
+        mock_log.assert_not_called()
+
+    def test_returns_empty_when_no_secret_arn(self, monkeypatch):
+        monkeypatch.delenv("LINEAR_API_TOKEN", raising=False)
+        monkeypatch.delenv("LINEAR_API_TOKEN_SECRET_ARN", raising=False)
+        assert resolve_linear_api_token() == ""
+
+    def test_import_error_degrades_gracefully(self, monkeypatch):
+        """If boto3 is missing from the container image, log WARN and return ''
+        rather than crashing the agent."""
+        monkeypatch.delenv("LINEAR_API_TOKEN", raising=False)
+        monkeypatch.setenv("LINEAR_API_TOKEN_SECRET_ARN", "arn:aws:sm:::secret/linear")
+        # Force `import boto3` (executed inside resolve_linear_api_token) to
+        # raise ImportError by removing it from sys.modules and shadowing it.
+        monkeypatch.setitem(sys.modules, "boto3", None)
+        with patch("config.log") as mock_log:
+            assert resolve_linear_api_token() == ""
+        # WARN logged, no exception escaped.
+        assert mock_log.call_count == 1
+        assert mock_log.call_args[0][0] == "WARN"
+        assert "boto3 unavailable" in mock_log.call_args[0][1]

agent/tests/test_linear_reactions.py

@@ -19,13 +19,17 @@
 
 
 @pytest.fixture(autouse=True)
-def _reset_viewer_cache():
-    """Reset the module-level viewer-id cache between tests so one test's
-    successful viewer fetch doesn't leak into another test that asserts the
-    sweep no-ops on viewer-fetch failure."""
+def _reset_module_state():
+    """Reset module-level caches and the auth circuit breaker between tests
+    so one test's state never leaks into another (viewer cache, consecutive
+    auth-failure counter, circuit-open flag)."""
     linear_reactions._viewer_id_cache = None
+    linear_reactions._consecutive_auth_failures = 0
+    linear_reactions._auth_circuit_open = False
     yield
     linear_reactions._viewer_id_cache = None
+    linear_reactions._consecutive_auth_failures = 0
+    linear_reactions._auth_circuit_open = False
 
 
 def _viewer_response(viewer_id: str = "viewer-bot") -> MagicMock:

cdk/src/handlers/orchestrate-task.ts

@@ -75,7 +75,16 @@ const durableHandler: DurableExecutionHandler<OrchestrateTaskEvent, void> = asyn
     if (!result) {
       await failTask(taskId, current.status, 'User concurrency limit reached', task.user_id, false);
       await emitTaskEvent(taskId, 'admission_rejected', { reason: 'concurrency_limit' });
-      await notifyLinearOnConcurrencyCap(task);
+      // Linear feedback is non-fatal: a throw here would re-run failTask +
+      // emitTaskEvent on the durable-execution retry, producing duplicate events.
+      try {
+        await notifyLinearOnConcurrencyCap(task);
+      } catch (err) {
+        logger.warn('Linear concurrency-cap feedback failed (non-fatal)', {
+          task_id: taskId,
+          error: err instanceof Error ? err.message : String(err),
+        });
+      }
     }
     return result;
   });

cdk/test/handlers/orchestrate-task-feedback.test.ts

@@ -128,4 +128,19 @@ describe('notifyLinearOnConcurrencyCap', () => {
       process.env.LINEAR_API_TOKEN_SECRET_ARN = saved;
     }
   });
+
+  test('reportIssueFailure rejection propagates (caller must catch)', async () => {
+    // The helper itself swallows network errors internally, but we contract
+    // for callers to wrap the call defensively because durable-execution
+    // retries the entire step on throw, producing duplicate failTask +
+    // emitTaskEvent. This test asserts the rejection actually propagates so
+    // the orchestrate-task try-catch is load-bearing, not redundant.
+    reportIssueFailureMock.mockRejectedValue(new Error('boom'));
+    await expect(
+      notifyLinearOnConcurrencyCap(task({
+        channel_source: 'linear',
+        channel_metadata: { linear_issue_id: 'lin-issue-1' },
+      })),
+    ).rejects.toThrow('boom');
+  });
 });