feat(ingestion): auto-extract awaiting-reply tasks from outbound emails

docs/testing/automated_test_catalog.md

@@ -61,15 +61,15 @@ flowchart TD
 - Do not hand-edit suite inventory entries in this file. Update the generator or the repository tree, then regenerate.
 
 ## Repo-wide summary
-- Total automated test files: **391**
-- Backend and repo Vitest files: **358**
+- Total automated test files: **392**
+- Backend and repo Vitest files: **359**
 - Frontend Vitest files: **9**
 - Playwright spec files: **24**
 
 ### Suite counts
 | Suite | Files |
 |---|---:|
-| Vitest unit tests | 96 |
+| Vitest unit tests | 97 |
 | Vitest service tests | 33 |
 | Source-adjacent tests | 45 |
 | Vitest integration tests | 106 |
@@ -107,7 +107,7 @@ flowchart TD
 **Runner:** `vitest`
 **Command:** `npm test -- tests/unit`
 **Requirements:** Basic `.env` if required by the module under test.
-**Files (96):**
+**Files (97):**
 - `tests/unit/aauth_admission.test.ts`
 - `tests/unit/aauth_attestation_apple_se.test.ts`
 - `tests/unit/aauth_attestation_revocation.test.ts`
@@ -190,6 +190,7 @@ flowchart TD
 - `tests/unit/sandbox_pack_registry.test.ts`
 - `tests/unit/sandbox_reset.test.ts`
 - `tests/unit/schema_agent_instructions.test.ts`
+- `tests/unit/schema_derived_entity_extraction.test.ts`
 - `tests/unit/schema_inference.test.ts`
 - `tests/unit/schema_projection_lag.test.ts`
 - `tests/unit/security_hardening.test.ts`

openapi.yaml

@@ -4686,6 +4686,13 @@ paths:
                   description: >
                     Entity IDs to link to this issue via REFERS_TO relationships.
                     Created server-side in the same operation as issue creation.
+                conversation_turn_id:
+                  type: string
+                  description: >
+                    Entity ID of the conversation turn (conversation_message entity) where this
+                    issue was observed. When provided, a REFERS_TO relationship is created from
+                    the filed issue entity to this conversation turn entity, making the origin
+                    of the issue traceable.
                 user_id:
                   type: string
       responses:
@@ -4784,6 +4791,61 @@ paths:
                 type: object
                 additionalProperties: true
 
+  /issues/import:
+    post:
+      summary: Import issues from JSONL
+      description: >
+        Bulk-import issues from a JSONL string (one JSON object per line) or a file path.
+        Each line is parsed as an issue object and ingested into the local Neotoma issue
+        entity graph. Intended for observer batch ingestion — importing issues exported
+        from another system without touching GitHub.
+        MCP import_issues_from_jsonl parity.
+      operationId: importIssuesFromJsonl
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: false
+              properties:
+                jsonl:
+                  type: string
+                  description: >
+                    Raw JSONL content — one JSON issue object per line.
+                    Mutually exclusive with file_path.
+                file_path:
+                  type: string
+                  description: >
+                    Absolute path to a JSONL file on the server's local filesystem.
+                    Mutually exclusive with jsonl.
+                user_id:
+                  type: string
+      responses:
+        "200":
+          description: Import summary
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties: false
+                required:
+                  - imported
+                  - skipped
+                  - errors
+                properties:
+                  imported:
+                    type: integer
+                    description: Number of issue objects successfully ingested.
+                  skipped:
+                    type: integer
+                    description: Number of lines skipped (blank or already-ingested with same idempotency key).
+                  errors:
+                    type: array
+                    items:
+                      type: string
+                    description: Per-line error messages for lines that failed to parse or ingest.
+
   /issues/sync:
     post:
       summary: Sync issues bidirectionally with GitHub

scripts/security/protected_routes_manifest.json

@@ -513,6 +513,18 @@
         401
       ]
     },
+    {
+      "path": "/issues/import",
+      "method": "POST",
+      "operation_id": "importIssuesFromJsonl",
+      "requires_auth": true,
+      "expected_no_auth_status": [
+        401
+      ],
+      "expected_invalid_auth_status": [
+        401
+      ]
+    },
     {
       "path": "/issues/status",
       "method": "POST",

src/actions.ts

@@ -132,6 +132,7 @@
   IssuesBulkEntityIdsRequestSchema,
   IssuesAddMessageRequestSchema,
   IssuesGetStatusRequestSchema,
+  IssuesImportFromJsonlRequestSchema,
   IssuesSubmitRequestSchema,
   IssuesSyncRequestSchema,
   EntitiesQueryRequestSchema,
@@ -1246,7 +1247,7 @@
       // Encryption off: local requests can use no-auth. HTTP (insecure) defaults to anonymous 000... user; HTTPS/localhost can use dev-local.
       if (!authHeader?.startsWith("Bearer ") && !connectionIdHeader) {
         if (isLocalRequest(req)) {
           const isInsecure = req.protocol === "http" || !(req as any).secure;
           if (isInsecure) {
             req.headers["x-connection-id"] = "dev-local-http";
             connectionIdHeader = "dev-local-http";
@@ -1502,7 +1503,7 @@
     await runWithRequestContext({ agentIdentity: fallbackIdentity, attributionDecision }, () =>
       transport!.handleRequest(req, res, req.body)
     );
   } catch (error: any) {
     logger.error("[MCP HTTP] Request error:", error);
     if (!res.headersSent) {
       res.status(500).json({
@@ -1865,7 +1866,7 @@
     const result = await initiateOAuthFlow(connection_id, client_name, finalRedirectUri);
 
     return res.json(result);
   } catch (error: any) {
     logError("MCPOAuthInitiate", req, error);
 
     // Check if it's a structured OAuthError
@@ -1875,7 +1876,7 @@
         message: string;
         statusCode: number;
         retryable?: boolean;
         details?: Record<string, any>;
       };
 
       return res.status(oauthError.statusCode).json({
@@ -1942,7 +1943,7 @@
       process.env.NEOTOMA_FRONTEND_URL || process.env.FRONTEND_URL || "http://localhost:5195";
     const successUrl = `${frontendBase}/oauth?connection_id=${encodeURIComponent(connectionId)}&status=success`;
     return res.redirect(successUrl);
   } catch (error: any) {
     logError("MCPOAuthCallback", req, error);
 
     // Extract structured error information
@@ -2239,7 +2240,7 @@
     });
 
     return res.redirect(result.authUrl);
   } catch (error: any) {
     logError("MCPOAuthAuthorize", req, error);
     return res.status(500).send(error.message ?? "Authorization failed");
   }
@@ -2329,7 +2330,7 @@
     return res.redirect(
       `${frontendOauth}?connection_id=${encodeURIComponent(connectionId)}&status=success`
     );
   } catch (error: any) {
     logError("MCPLocalLoginDevStub", req, error);
     const status = error?.code === "OAUTH_STATE_INVALID" || error?.statusCode === 400 ? 400 : 401;
     return res.status(status).send(
@@ -2406,7 +2407,7 @@
 
       res.setHeader("Content-Type", "application/json");
       return res.json(token);
     } catch (error: any) {
       logError("MCPOAuthToken", req, error);
       return res.status(400).json({
         error: "invalid_grant",
@@ -2434,7 +2435,7 @@
     });
     res.setHeader("Content-Type", "application/json");
     return res.status(201).json(reg);
   } catch (error: any) {
     logError("MCPOAuthRegister", req, error);
     if (error instanceof OAuthError) {
       const oauth = error as { code?: string; message?: string; statusCode?: number };
@@ -2470,7 +2471,7 @@
     const status = await getConnectionStatus(connection_id);
 
     return res.json({ status, connection_id });
   } catch (error: any) {
     logError("MCPOAuthStatus", req, error);
     return sendError(res, 500, "DB_QUERY_FAILED", error.message);
   }
@@ -6381,6 +6382,36 @@
           }`
         );
       }
+
+      // Schema-driven derived-entity extraction: if the entity's active schema
+      // declares `derived_entities`, evaluate each rule and create matching
+      // entities linked back to the source entity. Non-fatal: failures are
+      // logged and skipped, never blocking the primary store.
+      if (commit) {
+        try {
+          const { schemaRegistry: schemaReg2 } = await import("./services/schema_registry.js");
+          const schemaEntry2 = await schemaReg2.loadActiveSchema(r.entity_type, userId);
+          if (schemaEntry2?.schema_definition?.derived_entities?.length) {
+            const { extractDerivedEntities } =
+              await import("./services/schema_derived_entity_extraction.js");
+            await extractDerivedEntities({
+              entityId: r.entity_id,
+              entityType: r.entity_type,
+              fields: r.fields,
+              schema: schemaEntry2.schema_definition,
+              userId,
+              sourceId: observationSourceId,
+              idempotencyKey,
+            });
+          }
+        } catch (derivedErr) {
+          logger.warn(
+            `Derived entity extraction failed for ${r.entity_type}/${r.entity_id}: ${
+              derivedErr instanceof Error ? derivedErr.message : String(derivedErr)
+            }`
+          );
+        }
+      }
     }
 
     createdEntities.push({
@@ -8218,6 +8249,9 @@
           ...(parsed.data.entity_ids_to_link
             ? { entity_ids_to_link: parsed.data.entity_ids_to_link }
             : {}),
+          ...(parsed.data.conversation_turn_id
+            ? { conversation_turn_id: parsed.data.conversation_turn_id }
+            : {}),
         });
       })();
       logDebug("Success:issues_submit", req, { entity_id: result.entity_id });
@@ -8336,6 +8370,48 @@
 app.post("/issues/sync", handleIssuesSyncHttp);
 app.post("/api/issues/sync", handleIssuesSyncHttp);
 
+// POST /issues/import — JSONL batch ingestion (MCP import_issues_from_jsonl parity)
+const handleIssuesImportFromJsonlHttp: express.RequestHandler = async (req, res) => {
+  const parsed = IssuesImportFromJsonlRequestSchema.safeParse(req.body);
+  if (!parsed.success) {
+    logWarn("ValidationError:issues_import_from_jsonl", req, { issues: parsed.error.issues });
+    return sendValidationError(res, parsed.error.issues);
+  }
+  try {
+    const userId = await getAuthenticatedUserId(req, parsed.data.user_id);
+    const { createOperations } = await import("./core/operations.js");
+    const { NeotomaServer } = await import("./server.js");
+    const { importIssuesFromJsonl } = await import("./services/issues/import_from_jsonl.js");
+    const server = new NeotomaServer();
+    const ops = createOperations({ server, userId });
+    try {
+      const result = await importIssuesFromJsonl(ops, {
+        jsonl: parsed.data.jsonl,
+        filePath: parsed.data.file_path,
+      });
+      logDebug("Success:issues_import_from_jsonl", req, {
+        imported: result.imported,
+        skipped: result.skipped,
+        error_count: result.errors.length,
+      });
+      return res.json(result);
+    } finally {
+      await ops.dispose();
+    }
+  } catch (error) {
+    return handleApiError(
+      req,
+      res,
+      error,
+      "Failed to import issues from JSONL",
+      "ISSUES_IMPORT_FAILED",
+      "APIError:issues_import_from_jsonl"
+    );
+  }
+};
+app.post("/issues/import", handleIssuesImportFromJsonlHttp);
+app.post("/api/issues/import", handleIssuesImportFromJsonlHttp);
+
 // POST /restore_entity - Restore soft-deleted entity
 app.post("/restore_entity", async (req, res) => {
   const parsed = RestoreEntityRequestSchema.safeParse(req.body);

src/cli/index.ts

@@ -9085,6 +9085,24 @@ issuesCommand
     await issuesList({ ...opts, json: Boolean((program.opts() as { json?: boolean }).json) }, api);
   });
 
+issuesCommand
+  .command("import")
+  .description("Bulk-import issues from a JSONL file into local Neotoma")
+  .requiredOption("--from-jsonl <file>", "Path to a JSONL file (one issue JSON object per line)")
+  .action(async (opts) => {
+    const { issuesImport } = await import("./issues.js");
+    const config = await readConfig();
+    const token = await getCliToken();
+    const api = createApiClient({
+      baseUrl: await resolveBaseUrl(program.opts().baseUrl, config),
+      token,
+    });
+    await issuesImport(
+      { fromJsonl: opts.fromJsonl, json: Boolean((program.opts() as { json?: boolean }).json) },
+      api
+    );
+  });
+
 issuesCommand
   .command("sync")
   .description("Full sync of issues from GitHub into local Neotoma")

src/cli/issues.ts

@@ -64,6 +64,11 @@ export interface IssuesListOpts {
   json?: boolean;
 }
 
+export interface IssuesImportOpts {
+  fromJsonl: string;
+  json?: boolean;
+}
+
 export interface IssuesSyncOpts {
   since?: string;
   state?: "open" | "closed" | "all";
@@ -346,6 +351,52 @@ export async function issuesList(opts: IssuesListOpts, api: NeotomaApiClient): P
   process.stdout.write(`Use --no-sync to view cached local data only.\n`);
 }
 
+export async function issuesImport(opts: IssuesImportOpts, api: NeotomaApiClient): Promise<void> {
+  const { readFile } = await import("node:fs/promises");
+
+  let jsonl: string;
+  try {
+    jsonl = await readFile(opts.fromJsonl, "utf8");
+  } catch (err) {
+    process.stderr.write(
+      `issues import: failed to read file "${opts.fromJsonl}": ${(err as Error).message}\n`
+    );
+    process.exitCode = 1;
+    return;
+  }
+
+  const { data, error } = await api.POST("/issues/import", {
+    body: { jsonl },
+  });
+
+  if (error) {
+    process.stderr.write(`issues import failed: ${JSON.stringify(error)}\n`);
+    process.exitCode = 1;
+    return;
+  }
+
+  const row = data as
+    | {
+        imported?: number;
+        skipped?: number;
+        errors?: string[];
+      }
+    | undefined;
+
+  const imported = row?.imported ?? 0;
+  const skipped = row?.skipped ?? 0;
+  const errors = row?.errors ?? [];
+
+  if (opts.json) {
+    output({ imported, skipped, errors }, true);
+  } else {
+    process.stdout.write(`Imported ${imported} issues, skipped ${skipped}.\n`);
+    if (errors.length) {
+      process.stderr.write(`Errors:\n${errors.map((e) => `  - ${e}`).join("\n")}\n`);
+    }
+  }
+}
+
 export async function issuesSync(opts: IssuesSyncOpts, api: NeotomaApiClient): Promise<void> {
   const labelArr = opts.labels
     ? opts.labels

src/server.ts

@@ -978,6 +978,7 @@ export class NeotomaServer {
       reporter_ci_run_id: z.string().optional(),
       reporter_patch_source_id: z.string().optional(),
       entity_ids_to_link: z.array(z.string().min(1)).optional(),
+      conversation_turn_id: z.string().min(1).optional(),
     });
     const parsed = schema.parse(args ?? {});
 
@@ -1008,6 +1009,9 @@ export class NeotomaServer {
         reporter_ci_run_id: parsed.reporter_ci_run_id,
         reporter_patch_source_id: parsed.reporter_patch_source_id,
         ...(parsed.entity_ids_to_link ? { entity_ids_to_link: parsed.entity_ids_to_link } : {}),
+        ...(parsed.conversation_turn_id
+          ? { conversation_turn_id: parsed.conversation_turn_id }
+          : {}),
       });
 
       return this.buildTextResponse({