feat(noema-agent): add calendar conflict-check tool

.github/workflows/app-ci.yml

@@ -2,10 +2,6 @@ name: Application CI
 
 on:
   pull_request:
-    branches:
-      - develop
-      - master
-      - "release/**"
   push:
     branches:
       - develop
@@ -76,6 +72,21 @@ jobs:
             exit 1
           fi
 
+      - name: Run repository-root governance contract tests
+        # tests/ (repo root) holds workflow/YAML contract tests, e.g.
+        # test_stacked_pr_workflow_contract.py. The backend job's own pytest
+        # invocation above runs from backend/ and never collects this
+        # directory, so it needs its own explicit step. Mirrors that step's
+        # own log scan: a warning-class message pytest prints without
+        # actually failing the run must not be accepted as clean evidence.
+        run: |
+          set -o pipefail
+          python -m pytest -q tests 2>&1 | tee root_pytest_output.log
+          if grep -qiE 'timeout|fatal|warn|denied' root_pytest_output.log; then
+            echo "::error::Root governance contract tests produced Timeout, Fatal, Warn, or Denied outputs"
+            exit 1
+          fi
+
   frontend:
     name: frontend
     runs-on: ubuntu-latest

.github/workflows/bandit.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [ develop, master ]
   pull_request:
-    branches: [ develop, master ]
   workflow_dispatch:
 
 permissions:

.github/workflows/dependency-review.yml

@@ -2,10 +2,6 @@ name: Dependency Review
 
 on:
   pull_request:
-    branches:
-      - develop
-      - master
-      - "release/**"
   workflow_dispatch:
 
 permissions:

.github/workflows/docker-publish.yml

@@ -5,10 +5,6 @@ on:
     tags:
       - "v*"
   pull_request:
-    branches:
-      - develop
-      - master
-      - "release/**"
 
 permissions:
   contents: read

ARCHITECTURE.md

@@ -20,6 +20,21 @@ Runtime database connectivity is secret-injected: `backend/core/config.py` has
 no fallback `DATABASE_URL`, so missing database configuration fails at startup
 rather than silently using shared development credentials.
 
+## Reply follow-up scheduling boundary (Proposed, PR #1486)
+
+`ReplySlaScheduler` binds its existing ORM session to one checked-out connection
+for the entire PostgreSQL advisory-lease cycle, including per-workspace task
+commits and rollbacks. Normal release requires boolean confirmation. Acquisition
+uncertainty, cancellation, or failure invalidates that connection before session
+cleanup; a disconnected backend must not be replaced inside the same sweep.
+Healthy owner failures roll back and later owners are reloaded asynchronously.
+Manual escalation requests can race this scheduler: the shared service refreshes
+expired mail after a commit conflict and relies on savepoint rollback to remove
+failed inserts. The scheduler reloads owner configuration between workspaces.
+This retains source-linked task identity and the existing workspace queries.
+It does not promise exactly-once execution, provider writeback, or support for
+transaction-pooling proxies. See [decision and evidence](docs/doctoring/reply_sla_physical_lease.md).
+
 ## Topic-intelligence boundary
 
 Naruon has no live Structural Topic Modeling endpoint, fitted topic artifact,

backend/alembic/versions/0001_initial_control_plane.py

@@ -9,7 +9,7 @@
 from sqlalchemy import text
 
 from db.models import Base
-from scripts.bootstrap_db import schema_backfill_sql
+from scripts.bootstrap_db import execute_schema_backfill
 
 revision = "0001_initial_control_plane"
 down_revision = None
@@ -19,8 +19,7 @@ def upgrade() -> None:
     connection = op.get_bind()
     connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
     Base.metadata.create_all(connection)
-    for statement in schema_backfill_sql():
-        connection.execute(statement)
+    execute_schema_backfill(connection)
 
 
 def downgrade() -> None:

backend/alembic/versions/0011_email_read_state.py

@@ -1,29 +1,121 @@
 """Add is_read to emails (IMAP \\Seen read state).
 
 Existing rows default to read so historical/file imports do not surface as unread.
+
+Deliberate exception to this repo's "Alembic migrations use structured
+operations (``op.create_index``, ...), never ``sa.text(f"...")`` DDL" rule
+(``AGENTS.md``/``CLAUDE.md``): ``upgrade()``/``downgrade()`` below use
+``op.execute()`` with the module-level ``_UPGRADE_SQL``/``_DOWNGRADE_SQL``
+constants instead of a structured ``op.*`` call. That rule's actual target is
+DDL built from interpolated identifier strings (an injection-safety concern);
+these constants interpolate only ``_IS_READ_PROVENANCE_MARKER``, a fixed
+module-level literal, never an identifier or a value built from a variable,
+external input, or runtime state -- the same safety property a structured
+call would have. The reason a structured call isn't used is different: this
+migration's behavior must be conditional on whether the legacy ``emails``
+table exists, evaluated at apply time (see the comment on ``_UPGRADE_SQL``
+below for why that check cannot live in Python), and no structured Alembic
+operation expresses "run this DDL only if a runtime condition holds" -- a
+``DO $$ ... $$`` block is the correct primitive for that, not a workaround
+for one.
 """
 
 from alembic import op
-import sqlalchemy as sa
 
 # revision identifiers, used by Alembic.
 revision = "0011_email_read_state"
 down_revision = "0009_project_graph_projection"
 branch_labels = None
 depends_on = None
 
+# Fresh installations materialize the current ``email_records`` model in the
+# 0001 baseline, including ``is_read``. This historical side branch only
+# applies to databases that still carry its legacy ``emails`` table.
+#
+# The condition has to be evaluated in SQL, not Python: offline SQL
+# generation (``alembic upgrade --sql``, a real flag ``scripts/migrate_db.py``
+# exposes) has no live connection to introspect with and no specific target
+# database to ask "does this legacy table exist" at generation time either --
+# the same static script is meant to later be applied by a DBA against
+# whichever database they choose, fresh-install or legacy. A Python-side
+# check (``sa.inspect(op.get_bind())``) can only ever answer that question
+# for one hypothetical target chosen at generation time, so it is wrong for
+# the other: skip unconditionally and the column silently never gets added
+# for a legacy database that applies the generated script (while
+# ``alembic_version`` still advances, permanently hiding the gap); inspect
+# online and bake in one fixed answer and the same script fails outright
+# against the other kind of target. A ``DO $$ ... $$`` block defers the
+# check to apply time instead, so the one generated script is correct
+# against either kind of target, online or offline-then-applied-later alike.
+#
+# ``to_regclass('emails')`` (not ``information_schema.tables`` by bare
+# ``table_name``) deliberately: the unqualified ``ALTER TABLE emails`` below
+# resolves through the connection's ``search_path``, and ``to_regclass``
+# resolves an unqualified name exactly the same way, returning NULL if it
+# doesn't. ``information_schema.tables`` filtered only by ``table_name``
+# ignores ``search_path`` entirely and matches a same-named table in *any*
+# schema the connecting role can see -- on a deployment with more than one
+# accessible schema, that could find an unrelated ``emails`` table outside
+# the search path while the unqualified ``ALTER TABLE emails`` targets a
+# different (or no) table, passing the guard for the wrong relation or
+# aborting the migration outright. Resolving both the check and the DDL
+# through the same name lookup makes that mismatch structurally impossible.
+#
+# ``COMMENT ON COLUMN emails.is_read`` tags the column with a provenance
+# marker (``_IS_READ_PROVENANCE_MARKER``) the moment upgrade() actually adds
+# it. downgrade() only drops the column when that exact marker is present
+# (CodeRabbit, naruon#1501): an ``emails.is_read`` column that already
+# existed before this revision ran -- from some other, unrelated origin --
+# would upgrade()'s ``NOT EXISTS`` guard correctly leave alone, but an
+# unconditional ``DROP COLUMN IF EXISTS`` on downgrade would still destroy it
+# and its data, since a downgrade has no other way to tell "I added this"
+# apart from "this happens to be present". Checking the marker via
+# ``col_description`` makes downgrade drop only what this exact revision's
+# upgrade created.
+_IS_READ_PROVENANCE_MARKER = "0011_email_read_state:added"
+_UPGRADE_SQL = f"""
+DO $$
+BEGIN
+    IF to_regclass('emails') IS NOT NULL AND NOT EXISTS (
+        SELECT 1 FROM pg_attribute
+        WHERE attrelid = to_regclass('emails')
+        AND attname = 'is_read'
+        AND NOT attisdropped
+    ) THEN
+        ALTER TABLE emails ADD COLUMN is_read boolean NOT NULL DEFAULT true;
+        COMMENT ON COLUMN emails.is_read IS '{_IS_READ_PROVENANCE_MARKER}';
+    END IF;
+END $$;
+"""  # nosec B608
+
+_DOWNGRADE_SQL = f"""
+DO $$
+BEGIN
+    IF to_regclass('emails') IS NOT NULL AND EXISTS (
+        SELECT 1 FROM pg_attribute
+        WHERE attrelid = to_regclass('emails')
+        AND attname = 'is_read'
+        AND NOT attisdropped
+    ) AND col_description(to_regclass('emails'), (
+        SELECT attnum FROM pg_attribute
+        WHERE attrelid = to_regclass('emails')
+        AND attname = 'is_read'
+        AND NOT attisdropped
+    )) = '{_IS_READ_PROVENANCE_MARKER}' THEN
+        ALTER TABLE emails DROP COLUMN IF EXISTS is_read;
+    END IF;
+END $$;
+"""  # nosec B608
+
 
+# False positive on both calls below: _UPGRADE_SQL/_DOWNGRADE_SQL interpolate only the
+# fixed module-level literal _IS_READ_PROVENANCE_MARKER (see the module docstring above),
+# never external input or an identifier -- the same safety property a parameterized query
+# would have. Semgrep's raw-query/formatted-sql-query rules pattern-match on
+# "op.execute(f-string)" and cannot see that the interpolated value is a constant.
 def upgrade() -> None:
-    op.add_column(
-        "emails",
-        sa.Column(
-            "is_read",
-            sa.Boolean(),
-            nullable=False,
-            server_default=sa.text("true"),
-        ),
-    )
+    op.execute(_UPGRADE_SQL)  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query,python.lang.security.audit.formatted-sql-query.formatted-sql-query
 
 
 def downgrade() -> None:
-    op.drop_column("emails", "is_read")
+    op.execute(_DOWNGRADE_SQL)  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query,python.lang.security.audit.formatted-sql-query.formatted-sql-query

backend/alembic/versions/0018_calendar_conflict_judgments.py

@@ -0,0 +1,109 @@
+"""add calendar conflict judgments and corrections
+
+Revision ID: 0018_calendar_conflict_judgments
+Revises: 0017_merge_newsdom_carddav_heads
+Create Date: 2026-08-30 00:00:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0018_calendar_conflict_judgments"
+down_revision = "0017_merge_newsdom_carddav_heads"
+
+_JUDGMENT_TABLE = "calendar_conflict_judgments"
+_CORRECTION_TABLE = "calendar_conflict_corrections"
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+
+    if not inspector.has_table(_JUDGMENT_TABLE):
+        op.create_table(
+            _JUDGMENT_TABLE,
+            sa.Column("calendar_conflict_judgment_id", sa.Integer(), nullable=False),
+            sa.Column("judgment_uid", sa.String(length=96), nullable=False),
+            sa.Column("user_id", sa.String(), nullable=False),
+            sa.Column("organization_id", sa.String(), nullable=True),
+            sa.Column("workspace_id", sa.String(), nullable=False),
+            sa.Column("proposed_commitment_id", sa.String(length=256), nullable=False),
+            sa.Column("source_thread_id", sa.String(), nullable=True),
+            sa.Column("source_message_id", sa.String(), nullable=True),
+            sa.Column("decision_code", sa.String(length=32), nullable=False),
+            sa.Column("reason_code", sa.String(length=64), nullable=False),
+            sa.Column("recommended_action", sa.Text(), nullable=False),
+            sa.Column("policy_version", sa.String(length=32), nullable=False),
+            sa.Column("conflicts_json", sa.JSON(), nullable=False),
+            sa.Column("status_code", sa.String(length=32), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("calendar_conflict_judgment_id"),
+            sa.UniqueConstraint(
+                "judgment_uid", name="uq_calendar_conflict_judgments_uid"
+            ),
+        )
+
+    if not inspector.has_table(_CORRECTION_TABLE):
+        op.create_table(
+            _CORRECTION_TABLE,
+            sa.Column("calendar_conflict_correction_id", sa.Integer(), nullable=False),
+            sa.Column("correction_uid", sa.String(length=96), nullable=False),
+            sa.Column("calendar_conflict_judgment_id", sa.Integer(), nullable=False),
+            sa.Column("user_id", sa.String(), nullable=False),
+            sa.Column("organization_id", sa.String(), nullable=True),
+            sa.Column("workspace_id", sa.String(), nullable=False),
+            sa.Column("actor_user_id", sa.String(), nullable=False),
+            sa.Column("correction_action", sa.String(length=64), nullable=False),
+            sa.Column("before_json", sa.JSON(), nullable=False),
+            sa.Column("after_json", sa.JSON(), nullable=False),
+            sa.Column("rationale", sa.Text(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["calendar_conflict_judgment_id"],
+                ["calendar_conflict_judgments.calendar_conflict_judgment_id"],
+            ),
+            sa.PrimaryKeyConstraint("calendar_conflict_correction_id"),
+            sa.UniqueConstraint(
+                "correction_uid", name="uq_calendar_conflict_corrections_uid"
+            ),
+        )
+
+    for table_name, indexes in _calendar_conflict_indexes().items():
+        for index_name, column_names in indexes:
+            op.create_index(
+                index_name,
+                table_name,
+                column_names,
+                if_not_exists=True,
+            )
+
+
+def downgrade() -> None:
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+
+    for table_name in (_CORRECTION_TABLE, _JUDGMENT_TABLE):
+        if inspector.has_table(table_name):
+            for index_name, _column_names in reversed(
+                _calendar_conflict_indexes()[table_name]
+            ):
+                op.drop_index(index_name, table_name=table_name, if_exists=True)
+            op.drop_table(table_name)
+
+
+def _calendar_conflict_indexes() -> dict[str, list[tuple[str, list[str]]]]:
+    return {
+        _JUDGMENT_TABLE: [
+            (
+                "ix_calendar_conflict_judgments_scope_thread",
+                ["user_id", "organization_id", "workspace_id", "source_thread_id"],
+            ),
+        ],
+        _CORRECTION_TABLE: [
+            (
+                "ix_calendar_conflict_corrections_judgment",
+                ["calendar_conflict_judgment_id"],
+            ),
+        ],
+    }