feat(integrations): reuse Engine reviews and durable chat admission [skip ci]

GitHub Action exports validated structured findings for a separate trusted publisher with current-head checks and dedupe. Telegram persists prompt identity before cursor advancement, reconciles the same Engine operation and retains uncertain final replies for explicit status recovery. DingTalk internal-app Stream adapter reuses bridge-core admission, approvals and question parsing. Extracted existing Weixin question helpers into shared core; no second turn loop.

Local evidence:168 existing bridge/Telegram/Weixin/Feishu/WeCom checks pass,1existing skip;9 actual local HTTP bridge-process scenarios pass including restart/identity/question/uncertain-final delivery. Action8 existing tests plus30 injected publisher/contract scenarios pass. DingTalk20 injected transport scenarios pass plus syntax/SDK import; no live connection. Feature derivation/whitespace pass. Staged shared registry/changelog/facts exclude concurrent instruction-budget work. No new tests/comments, native build, provider completion, workflow activation or deploy. Refs CW-6864.
This commit is contained in:
CodeWhale Bot committed 2026-10-26 00:30:04 -07:00
1 parent fb10884b4f
commit 81558024ca
25 files changed
+2005 -339

No files matched your search

+4
View File
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
- **GitHub Action reviews:** analysis can export a structured Engine review for a separate trusted publisher job. The publisher checks the current PR revision and line anchors, deduplicates its findings, and keeps uncertain or preexisting claims out of inline comments.
- **Telegram recovery:** prompts and their initiating identity are durable before the polling cursor advances. Restart recovery reuses the original Engine operation, questions preserve human ownership, and uncertain final replies remain available through `/status` without an automatic resend.
- **DingTalk bridge:** an optional internal-app Stream adapter reuses the same Engine, shared admission and approval helpers. Tenant/staff allowlists, durable acceptance before acknowledgement and retained uncertain deliveries protect recovery. Managed account linking is separate from this advanced self-hosted bridge.
- **Answerless output-limit recovery:** opt into `[reasoning_only] on_output_limit = true` to request an answer again after a length stop with no answer or tool call, including hidden reasoning and ordinary Fleet workers. Retries share `max_reprompts`, retain existing output and control limits, and record request-scoped nudge receipts without adding them to resumed conversation context. RLM keeps its existing stop protocol. Thanks [@M-Maciej](https://github.com/M-Maciej) for the long-proof Fleet report. (#6990)
- **Linux hardening tests:** the NoNewPrivs child matrices now respect a parent that already holds the irreversible flag, keeping the full assertions on unpinned hosts and checking the kernel ratchet on pinned hosts. Thanks [@asto18089](https://github.com/asto18089) for diagnosing the hardened-container failure. (#7021)
+5 -2
View File
@@ -11,7 +11,7 @@ inputs:
description: Exact model ID. Account mode requires provider/model from the account catalog.
required: true
github-token:
description: GitHub token with contents read and pull-requests write.
description: GitHub token with contents read and pull-requests read; legacy post:true also requires pull-requests write.
default: ${{ github.token }}
pr-number:
description: PR number for workflow_dispatch; pull_request events use the event snapshot.
@@ -27,7 +27,7 @@ inputs:
description: Model and publication deadline, in seconds (30–1200).
default: '600'
post:
description: Publish a COMMENT review; never approve or request changes.
description: Legacy direct COMMENT publication. Set false to export findings for a separate publisher job.
default: 'true'
outputs:
outcome:
@@ -36,6 +36,9 @@ outputs:
receipt:
description: Absolute path to the sanitized JSON execution receipt (also written on failures).
value: ${{ steps.review.outputs.receipt }}
findings:
description: Absolute path to validated findings JSON for a complete post:false review; contains repository-sensitive review text.
value: ${{ steps.review.outputs.findings }}
pr-url:
description: URL of the pull request, when resolved.
value: ${{ steps.review.outputs.pr-url }}
+145 -66
View File
@@ -2,22 +2,35 @@
The root `action.yml` runs the existing `codewhale review` command with an
exact, checksummed release. It needs Node.js 22+, `gh`, Git, and a macOS or
Linux runner. It does not compile Codewhale or execute the PR's code.
Linux runner. It fetches Git objects without checking out or executing PR code.
The Engine owns review coverage, prompts, routing and finding generation.
Use two jobs: analysis with a read-only GitHub token and the selected provider
key, followed by publication with a write token and no provider key. Analysis
exports a validated findings artifact. The publisher validates its repository,
PR, workflow run and attempt, and current base/head before submitting a
`COMMENT` review. It never approves, requests changes, commits, or runs a model.
Review mode only. Autonomous mention/fix mode is not implemented in this
Action. The Action has been in release tags since `v0.10.1`; pin a tag or a
reviewed commit. The CLI `version` input and the Action ref are separate pins.
Action. The original direct-review Action has been in release tags since
`v0.10.1`; the findings output and separate publisher require a commit containing
both features. The Action ref and the exact released CLI `version` are separate
pins. Adding these source files does not enable a workflow.
## Setup with your own provider key
Pick a provider (`deepseek`, `anthropic`, `openrouter`, `zai`, or
`modelstudio-token-plan`) and an exact model ID. Save that provider's key as a
repository Actions secret named for it: `DEEPSEEK_API_KEY`,
`ANTHROPIC_API_KEY`, `OPENROUTER_API_KEY`, `ZAI_API_KEY` or
`MODELSTUDIO_API_KEY`. Pass only that one secret in `env`. The Action never
chooses a different credential after an authentication or payment failure.
Pick `deepseek`, `anthropic`, `openrouter`, `zai`, or `modelstudio-token-plan`,
an exact model ID, and a released CLI tag. Store the matching provider key as
an Actions secret. Pass only that provider's key to the analysis job. The
Action never chooses a different credential after an authentication or payment
failure. The example uses DeepSeek; replace its provider and secret together
when choosing another route.
Add `.github/workflows/codewhale.yml`:
Set `CODEWHALE_REVIEW_VERSION` to an exact `vX.Y.Z` release and
`CODEWHALE_REVIEW_MODEL` to the chosen provider's exact model ID. Replace both
`REPLACE_WITH_REVIEWED_COMMIT_SHA` placeholders with the same reviewed Action
commit containing the publisher. Pin third-party actions to reviewed commits
under your repository's Actions policy.
```yaml
name: Codewhale review
@@ -30,59 +43,101 @@ on:
description: Same-repository PR number
required: true
type: string
permissions:
contents: read
pull-requests: write
permissions: {}
concurrency:
group: codewhale-review-${{ github.event.pull_request.number || inputs.pr-number }}
cancel-in-progress: true
cancel-in-progress: false
jobs:
review:
runs-on: ubuntu-latest
timeout-minutes: 25
permissions:
contents: read
pull-requests: read
outputs:
findings: ${{ steps.review.outputs.findings }}
steps:
- uses: actions/setup-node@v7
with:
node-version: '22'
- uses: codewhale-hq/CodeWhale@v0.10.1
- uses: codewhale-hq/CodeWhale@REPLACE_WITH_REVIEWED_COMMIT_SHA
id: review
with:
version: v0.10.1
version: ${{ vars.CODEWHALE_REVIEW_VERSION }}
provider: deepseek
model: deepseek-flash
model: ${{ vars.CODEWHALE_REVIEW_MODEL }}
pr-number: ${{ inputs.pr-number }}
post: 'false'
env:
DEEPSEEK_API_KEY: ${{ (github.event_name == 'workflow_dispatch' || github.event.pull_request.head.repo.full_name == github.repository) && secrets.DEEPSEEK_API_KEY || '' }}
- uses: actions/upload-artifact@v7
if: steps.review.outputs.findings != ''
with:
name: codewhale-findings-${{ github.run_id }}-${{ github.run_attempt }}
path: ${{ steps.review.outputs.findings }}
retention-days: 1
- uses: actions/upload-artifact@v7
if: always() && steps.review.outputs.receipt != ''
with:
name: codewhale-review-${{ github.run_id }}-${{ github.run_attempt }}
name: codewhale-review-receipt-${{ github.run_id }}-${{ github.run_attempt }}
path: ${{ steps.review.outputs.receipt }}
retention-days: 14
publish:
needs: review
if: needs.review.outputs.findings != ''
runs-on: ubuntu-latest
permissions:
contents: read
pull-requests: write
steps:
- uses: actions/setup-node@v7
with:
node-version: '22'
- uses: actions/download-artifact@v7
with:
name: codewhale-findings-${{ github.run_id }}-${{ github.run_attempt }}
path: findings
- uses: codewhale-hq/CodeWhale/scripts/github-action/publish@REPLACE_WITH_REVIEWED_COMMIT_SHA
id: publish
with:
findings: findings/findings.json
pr-number: ${{ inputs.pr-number }}
- uses: actions/upload-artifact@v7
if: always() && steps.publish.outputs.receipt != ''
with:
name: codewhale-publication-receipt-${{ github.run_id }}-${{ github.run_attempt }}
path: ${{ steps.publish.outputs.receipt }}
retention-days: 14
```
Always set `provider`: its default, `codewhale`, is the account relay below.
No repository checkout is needed in either job. The publishing job must use
trusted Action code and must not run files from the findings artifact. Keep it
on a separate ephemeral runner; a later step on the analysis runner is not the
same credential boundary. The analysis CLI still receives a read-only GitHub
token for PR metadata and Git objects, alongside its selected provider key.
### Account machine key: not available
Never use `pull_request_target` with these actions. Fork PRs and drafts are
ineligible even on a manual run. A skipped event is not a clean review. This
publisher accepts artifacts only from the current workflow run and attempt;
cross-workflow `workflow_run` publication is not implemented.
The `codewhale` provider routes through a Codewhale account machine key
(`CODEWHALE_API_KEY`, with `model` as `provider/model` from the account
catalog). Codewhale accounts are paused for new users while the account and
key-storage paths get a security review, so use your own provider key.
To publish as a dedicated GitHub App, create an installation token only in the
publisher job, pass it as `github-token`, and set `reviewer-login` to that App's
exact bot login, such as `codewhale-agent[bot]`. The publisher verifies the
returned review author; a mismatch is an uncertain publication, not permission
to retry. See [App identity setup](GITHUB_APP.md).
No repository checkout is needed. To post as your own GitHub App, supply an
installation token as `github-token`; see [App identity setup](GITHUB_APP.md).
Never use `pull_request_target` with this Action. Fork PRs and drafts are
ineligible even on a manual run. A skipped event is not a clean review.
### Account machine keys
For BYOK, set `provider` explicitly (`deepseek`, `anthropic`, `openrouter`,
`zai`, or `modelstudio-token-plan`), set its exact `model`, and pass only the
matching provider secret in `env`. Omit `CODEWHALE_API_KEY`. The Action never
chooses a different credential after an authentication or payment failure.
The `codewhale` provider uses `CODEWHALE_API_KEY` and an exact `provider/model`
account-catalog entry. Account sign-up availability does not establish that
machine-key relay access is qualified for Actions. Use explicit BYOK unless
your account's machine-key route has been separately verified. The Action runs
`codewhale account agent` before using that route.
## Bounds and evidence
| Input | Default | Meaning |
| Analysis input | Default | Meaning |
| --- | --- | --- |
| `version` | Required | Exact released CLI tag, never `latest` |
| `provider` | `codewhale` | Account relay or explicit BYOK |
@@ -91,46 +146,70 @@ chooses a different credential after an authentication or payment failure.
| `max-passes` | 1 | Complete ordered passes, maximum 64 |
| `max-output-tokens` | CLI automatic | Optional per-pass ceiling, 8192–1000000 |
| `timeout-seconds` | 600 | Model/publication deadline, 30–1200 |
| `post` | `true` | `false` produces a receipt without publishing |
| `post` | `true` | Legacy direct publication; set `false` for the two-job setup |
These are input, output, and time bounds, not a dollar guarantee. Model prices
and reasoning accounting vary. Raising the pass count authorizes more model
requests. The Action never retries inference automatically. Do not enable it
without setting the desired review scope and spend limits with your provider.
These bounds do not guarantee a dollar cost. Model prices and reasoning
accounting vary. Raising the pass count authorizes more model requests.
Inference is never automatically retried. Set the intended review scope and
provider spend controls before enabling a workflow.
Outputs are `outcome`, `receipt` (absolute JSON path), and `pr-url`. The
receipt contains the pinned revision, route, limits, publication state,
completion counts and reported token usage. It excludes raw model output,
provider errors, PR text and credentials. Provider token accounting can be
absent. A runner shutdown before the receipt is written has no completion
receipt and must not be counted as a clean review.
Analysis outputs are `outcome`, `receipt`, `findings`, and `pr-url`. The
sanitized receipt records revision, route, bounds, completion counts,
publication state, and reported token usage. It excludes raw model/provider
output, PR text and credentials. Missing token accounting stays missing.
A runner shutdown before receipt creation is not a completed review.
The CLI validates complete diff coverage and checks current revision before
publication. It reads bounded source excerpts from pinned Git blobs, without
running tests or investigating arbitrary unchanged callers. “Reviewed clean”
means the configured review completed with zero reported issues; it is not
proof that the PR contains no bugs.
`findings` is populated only for complete `post:false` reviews. It contains
validated structured review text, evidence, file locations and exact revision
and workflow identity. Unknown fields, raw model output, provider errors and
replacement code are excluded; known supplied credentials are redacted. It
can still contain private source excerpts or other repository-sensitive text.
Store it only with the repository's intended artifact access and retention.
The publisher reads the complete current PR file list and existing reviews,
then rechecks the base/head. It anchors inline findings only where GitHub
supplies a complete valid patch containing their line. Pre-existing,
unanchorable and suspected findings remain in the summary. Suggestions are
prose only, with no one-click replacement blocks. Up to 50 findings are also
posted inline; every retained finding stays in the summary. Oversized review
bodies fail before any write instead of silently dropping findings.
The publisher recognizes its own review/finding fingerprints only on reviews
by `reviewer-login`. Identical review reruns are skipped. Findings with matching
normalized title/description, severity, location and unchanged file content are not repeated,
including after unrelated files change. Changed file contents can produce a
fresh finding. This is deterministic deduplication, not semantic similarity
judgment; differently worded findings can still recur. Concurrency must remain
serialized per PR because GitHub does not offer an atomic idempotency key for
review creation.
The CLI validates complete diff coverage. It reads bounded excerpts from
pinned Git blobs; it does not execute PR tests or inspect arbitrary unchanged
callers. “Reviewed clean” means zero issues were reported in a completed
review, not proof that the PR contains no bugs.
## Outcomes and recovery
| Outcome | Meaning / next action |
| --- | --- |
| `reviewed_clean` | Complete model review with zero reported issues |
| `reviewed_with_findings` | Complete model review; findings remain advisory |
| `configuration_missing` | Check the exact release/model, key presence, route and input bounds |
| `failed` | No complete review receipt; check account readiness, provider balance, release assets and GitHub access |
| `incomplete` | Coverage was incomplete; inspect the PR and revise scope or limits |
| `publication_uncertain` | Check the PR before any retry; publication may have succeeded |
| `superseded` | PR revision changed; run against the current head |
| `not_eligible` | Fork, draft, closed PR or unsupported event; no model run |
| `reviewed_clean` | Complete analysis with zero reported issues |
| `reviewed_with_findings` | Complete analysis; findings remain advisory |
| `published` | Publisher confirmed a `COMMENT` review at the captured head |
| `already_published` | Matching review or findings already exist under the configured publisher |
| `configuration_missing` | Check release/model, keys, bounds, artifact and workflow identity |
| `failed` | No complete analysis/publication receipt; inspect the safe reason code |
| `incomplete` | Coverage incomplete; revise scope or limits |
| `publication_uncertain` | Check the PR before retrying; the write may have succeeded |
| `superseded` | Base or head changed; analyze the current revision |
| `not_eligible` | Fork, draft, closed PR or unsupported event; no model run/write |
Failures fail the optional Actions job; do not make it a required merge check
unless that is your repository policy. A provider failure is not a negative
verdict about the PR. No additional failure comment is posted.
Failures fail the optional job; whether it is a required merge check is the
repository owner's policy. A provider failure is not a negative verdict about
the PR. No additional failure comment is posted.
After repairing setup, use “Run workflow” with a PR number. Enabling a disabled
workflow does not replay old events. Manual reruns can publish another review:
cross-run publication deduplication is not implemented in this Action. Check
GitHub first, especially after a timeout or cancellation. Recovery is proved
by a complete receipt for the current head plus the intended publication,
not by workflow enablement alone.
After repairing setup, run the entire workflow with a PR number. Rerunning
only publication in a new run attempt rejects the old artifact; this prevents
silently publishing a bundle from another execution. Enabling a disabled
workflow does not replay old events. Legacy `post:true` preserves the CLI's
direct publication behavior and does not receive the separate publisher's
cross-run deduplication or write-token isolation.
+30
View File
@@ -808,3 +808,33 @@ status = "stable"
since = "0.10.2"
surfaces = { tui = "stable" }
owner = "crates/tui/src/tui/herdr.rs"
[[feature]]
id = "github-action-review"
name = "GitHub Action review"
summary = "Review pull requests with the Engine and publish validated findings from a separate trusted job."
status = "experimental"
since = "unreleased"
surfaces = { tui = "preview" }
docs = "docs/GITHUB_ACTION.md"
owner = "scripts/github-action/review.mjs"
[[feature]]
id = "telegram-agent"
name = "Telegram agent bridge"
summary = "Use an Engine from Telegram with durable prompt recovery, human-owned approvals and retained replies."
status = "experimental"
since = "unreleased"
surfaces = { api = "preview" }
docs = "integrations/telegram-bridge/README.md"
owner = "integrations/telegram-bridge/src/index.mjs"
[[feature]]
id = "dingtalk-agent"
name = "DingTalk agent bridge"
summary = "Connect an internal DingTalk app to the same Engine with scoped identity and durable message recovery."
status = "experimental"
since = "unreleased"
surfaces = { api = "preview" }
docs = "integrations/dingtalk-bridge/README.md"
owner = "integrations/dingtalk-bridge/src/index.mjs"
+101 -6
View File
@@ -66,7 +66,7 @@ export async function writeFileDurable(filePath, contents, { mode = 0o600 } = {}
export class ThreadStore {
static async open(filePath, options = {}) {
const store = new ThreadStore(filePath, options);
const store = new this(filePath, options);
await store.load();
return store;
}
@@ -146,12 +146,17 @@ export class ThreadStore {
await this.save();
}
async recordMessage(messageKey) {
if (!messageKey || this.options.messageLimit <= 0) return false;
async recordMessage(messageKey, chatId = null, patch = null) {
if (patch && !chatId) throw new Error("Durable admission requires a chat binding");
if (!messageKey || this.options.messageLimit <= 0) {
if (patch) throw new Error("Durable admission requires message retention");
return false;
}
this.ensureShape();
if (this.data.messages.includes(messageKey)) return true;
this.data.messages.push(messageKey);
this.data.messages = this.data.messages.slice(-this.options.messageLimit);
if (chatId && patch) this.data.chats[chatId] = { ...this.data.chats[chatId], ...patch };
await this.save();
return false;
}
@@ -636,7 +641,7 @@ export async function* readSse(response) {
}
}
export function createRuntimeClient({ runtimeUrl, runtimeToken }) {
export function createRuntimeClient({ runtimeUrl, runtimeToken, runtimeRequestTimeoutMs }) {
function authHeaders() {
return { authorization: `Bearer ${runtimeToken}` };
}
@@ -648,11 +653,14 @@ export function createRuntimeClient({ runtimeUrl, runtimeToken }) {
...(options.auth === false ? {} : authHeaders()),
...(options.body ? { "content-type": "application/json" } : {})
},
body: options.body ? JSON.stringify(options.body) : undefined
body: options.body ? JSON.stringify(options.body) : undefined,
signal: options.signal || (runtimeRequestTimeoutMs ? AbortSignal.timeout(runtimeRequestTimeoutMs) : undefined)
});
const body = await readJsonSafe(response);
if (!response.ok) {
throw new Error(compactRuntimeError(response.status, body));
const error = new Error(compactRuntimeError(response.status, body));
error.status = response.status;
throw error;
}
return body;
}
@@ -668,6 +676,93 @@ export function compactRuntimeError(status, body) {
return `Runtime API request failed (${status}): ${message}`;
}
export function questionText(request) {
const questions = Array.isArray(request?.questions) ? request.questions : [];
const lines = ["需要你回答 / Codewhale is asking"];
for (const [index, question] of questions.entries()) {
const prefix = questions.length > 1 ? `${index + 1}. ` : "";
lines.push(`${prefix}${question.header ? `${question.header}: ` : ""}${question.question || ""}`);
for (const [optionIndex, option] of (question.options || []).entries()) {
const description = String(option.description || "").trim();
lines.push(` ${optionIndex + 1}) ${option.label}${option.description ? ` - ${description.length > 120 ? `${description.slice(0, 119)}…` : description}` : ""}`);
}
}
lines.push(questions.length > 1
? "每行回答一个问题,可回复选项编号或文字。/ Answer one question per line, by option number or in words."
: "回复选项编号或文字。/ Reply with an option number or in words.");
return lines.join("\n");
}
export function parseQuestionAnswers(request, text) {
const questions = Array.isArray(request?.questions) ? request.questions : [];
if (!questions.length) return null;
const replies = questions.length > 1
? String(text || "").split(/\n+/).map((line) => line.trim()).filter(Boolean)
: [String(text || "").trim()];
if (replies.length !== questions.length || replies.some((reply) => !reply)) return null;
const answers = [];
for (const [index, question] of questions.entries()) {
const reply = replies[index];
const options = question.options || [];
const picks = question.multi_select ? reply.split(/[\s,,、]+/u).filter(Boolean) : [reply];
const chosen = picks.every((pick) => /^\d+$/.test(pick) && Number(pick) >= 1 && Number(pick) <= options.length)
? picks.map((pick) => options[Number(pick) - 1])
: options.filter((option) => option.label.toLowerCase() === reply.toLowerCase());
if (chosen.length) {
for (const option of chosen) answers.push({ id: question.id, label: option.label, value: option.label });
} else if (question.allow_free_text || !options.length) {
answers.push({ id: question.id, label: "Other", value: reply });
} else {
return null;
}
}
return answers;
}
export async function admitTurn({ runtimeJson, pending, capabilities, persistPending, ensureThread }) {
if (!pending?.request || (pending.operationKey && pending.request.operation_key !== pending.operationKey)) {
throw new Error("Invalid retained turn admission");
}
if (pending.operationKey && (capabilities?.turn_operation_idempotency !== true || capabilities?.turn_operation_lookup !== true)) {
return { pending, turn: null, uncertain: true };
}
if (!pending.threadId) {
const threadId = await ensureThread(pending.threadRequest);
if (!threadId) throw new Error("Turn admission requires a durable thread binding");
pending = { ...pending, threadId };
await persistPending(pending);
}
if (pending.sinceSeq === null) {
const detail = await runtimeJson(`/v1/threads/${encodeURIComponent(pending.threadId)}`);
if (latestRunningTurn(detail)) return { pending, turn: null, blocked: true };
pending = { ...pending, sinceSeq: Number(detail.latest_seq || 0) };
await persistPending(pending);
}
let turn;
if (pending.submitted) {
if (!pending.operationKey || capabilities?.turn_operation_idempotency !== true || capabilities?.turn_operation_lookup !== true) {
return { pending, turn: null, uncertain: true };
}
try {
turn = await runtimeJson(`/v1/threads/${encodeURIComponent(pending.threadId)}/turn-operations/${encodeURIComponent(pending.operationKey)}`);
} catch (error) {
if (error.status !== 404) throw error;
const detail = await runtimeJson(`/v1/threads/${encodeURIComponent(pending.threadId)}`);
if (latestRunningTurn(detail)) return { pending, turn: null, blocked: true };
}
}
if (!turn) {
pending = { ...pending, submitted: true };
await persistPending(pending);
const result = await runtimeJson(`/v1/threads/${encodeURIComponent(pending.threadId)}/turns`, {
method: "POST", body: pending.request
});
turn = result?.turn;
}
if (!turn?.id || turn.thread_id !== pending.threadId) throw new Error("Invalid Engine admission receipt");
return { pending, turn };
}
export function latestRunningTurn(detail) {
const turns = Array.isArray(detail?.turns) ? detail.turns : [];
for (let index = turns.length - 1; index >= 0; index -= 1) {
@@ -1,4 +1,5 @@
import test from "node:test";
import { createHash } from "node:crypto";
import { createServer } from "node:http";
import { once } from "node:events";
import assert from "node:assert/strict";
@@ -8,8 +9,9 @@ import path from "node:path";
import url from "node:url";
import {
ApprovalOwnershipError, ThreadStore, decideApproval, settleStoredApproval,
createRuntimeClient, activeTurnBlock, readSse, readJsonSafe, compactRuntimeError
admitTurn, createRuntimeClient, activeTurnBlock, latestRunningTurn, readSse, readJsonSafe, compactRuntimeError
} from "../src/lib.mjs";
import { isAllowed, isGroupChat } from "../../telegram-bridge/src/lib.mjs";
import { incomingIdentity as wecomIdentity } from "../../wecom-bridge/src/lib.mjs";
const root = path.resolve(path.dirname(url.fileURLToPath(import.meta.url)), "..", "..");
@@ -28,7 +30,7 @@ const actors = {
async function withStore(fn, options = {}) {
const dir = await mkdtemp(path.join(tmpdir(), "bridge-approvals-"));
try {
const store = await ThreadStore.open(path.join(dir, "state.json"), { actions: true, ...options });
const store = await ThreadStore.open(path.join(dir, "state.json"), { actions: true, messageLimit: 500, ...options });
await store.setChat(chatId, { threadId });
await fn(store);
} finally { await rm(dir, { recursive: true, force: true }); }
@@ -43,13 +45,17 @@ function extract(source, name) {
}
async function shipping(bridge, names, values) {
const source = await readFile(path.join(root, bridge, "src/index.mjs"), "utf8");
const functions = bridge === "weixin-bridge" && names.includes("turnActor") ? ["accountIdentity", ...names] : names;
const functions = [...new Set([
...(bridge === "weixin-bridge" && names.includes("turnActor") ? ["accountIdentity"] : []),
...(bridge === "telegram-bridge" && (names.includes("runPrompt") || names.includes("startPromptTurn"))
? ["recoveryAllowed", "supportsAdmissionRecovery"] : []), ...names
])];
return new Function(...Object.keys(values), functions.map((name) => extract(source, name)).join("\n") +
`\nreturn { ${names.join(", ")} };`)(...Object.values(values));
}
async function withRuntime(fn) {
const state = { pending: [{ id: "approval-a", turn_id: turnId }], turns: [], posts: [],
startAttempts: 0, accepted: [], failStart: false, acceptDecision: true, events: [] };
startAttempts: 0, accepted: [], operations: new Map(), failStart: false, acceptDecision: true, events: [] };
const server = createServer(async (request, response) => {
assert.equal(request.headers.authorization, "Bearer bridge-runtime-fixture-token");
let text = "";
@@ -69,8 +75,15 @@ async function withRuntime(fn) {
} else {
const id = `accepted-${state.startAttempts}`;
state.accepted.push(id);
response.end(JSON.stringify({ turn: { id } }));
const turn = { id, thread_id: threadId };
const operationKey = JSON.parse(text).operation_key;
if (operationKey) state.operations.set(operationKey, turn);
response.end(JSON.stringify({ turn }));
}
} else if (request.url.includes("/turn-operations/")) {
const turn = state.operations.get(decodeURIComponent(request.url.split("/").at(-1)));
response.statusCode = turn ? 200 : 404;
response.end(JSON.stringify(turn || { error: { message: "operation not found" } }));
} else if (request.method === "POST") {
state.posts.push({ route: request.url, body: JSON.parse(text) });
response.statusCode = state.acceptDecision ? 200 : 503;
@@ -92,19 +105,27 @@ function environment(store, runtime, extra = {}) {
settleStoredApproval, ApprovalOwnershipError, incomingIdentity: wecomIdentity,
sendText: async () => {}, replyText: async () => {}, sendTurnText: async () => {},
parseApprovalDecisionArgs: () => { throw new Error("fixture uses actual parsed approval id"); },
ensureThread: async () => store.getChat(chatId), activeTurnBlock,
config: { runtimeUrl: runtime.runtimeUrl, turnTimeoutMs: 1000, maxReplyChars: 10000 },
ensureThread: async () => store.getChat(chatId), activeTurnBlock, latestRunningTurn,
admitTurn, createHash, isAllowed, isGroupChat, botAccountId: "fixture-bot", activeTurnTasks: new Map(),
runtimeCapabilities: { turn_operation_idempotency: true, turn_operation_lookup: true },
startTrackedTurnStream: () => {}, delay: (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
config: { runtimeUrl: runtime.runtimeUrl, turnTimeoutMs: 1000, maxReplyChars: 10000, allowGroups: true, allowUnlisted: true, allowlist: [] },
controlKeyboard: () => ({}), activeTurnKeyboard: () => ({}), helpText: () => "",
clearActiveTurn: () => store.patchChat(chatId, { activeTurnId: null }), stopping: false,
streamTurnEvents: async (chat, thread, turn) => {
const reopened = await ThreadStore.open(store.filePath, { actions: true });
assert.ok(reopened.turnOrigin(chat, thread, turn), "origin must be durable before delivery");
return true;
}, ...extra
};
}
async function invokeStart(bridge, fn, user = "alice") {
if (bridge === "telegram-bridge") return fn(chatId, "prompt", { actorId: `telegram-user:${user}` });
return fn(chatId, "prompt", identities[bridge](user));
let incomingNumber = 0;
async function invokeStart(bridge, functions, user = "alice") {
if (bridge === "telegram-bridge") {
await functions.startPromptTurn(chatId, "prompt", { ...identities[bridge](user), messageKey: `message-${++incomingNumber}` });
return functions.runPrompt(chatId);
}
return functions.runPrompt(chatId, "prompt", identities[bridge](user));
}
test("approval tokens retain 128-bit entropy and require their accepted human and turn", async () => {
@@ -246,27 +267,27 @@ test("Weixin shipping decision refuses a legacy origin without bot account prove
for (const bridge of ["telegram-bridge", "feishu-bridge", "wecom-bridge"]) {
test(`${bridge} accepted start binds before delivery; failed, queued and later starts cannot transfer an old approval`, async () => {
await withRuntime(async (runtime) => withStore(async (store) => {
const functions = await shipping(bridge, ["turnActor", "runPrompt", "decideApproval"], environment(store, runtime, bridge === "wecom-bridge" ? {
const functions = await shipping(bridge, ["turnActor", "runPrompt", "decideApproval", ...(bridge === "telegram-bridge" ? ["startPromptTurn"] : [])], environment(store, runtime, bridge === "wecom-bridge" ? {
streamTurnEvents: async (chat, frame, thread, turn) => {
const reopened = await ThreadStore.open(store.filePath, { actions: true });
assert.ok(reopened.turnOrigin(chat, thread, turn), "origin must be durable before delivery");
}
} : {}));
await invokeStart(bridge, functions.runPrompt, "alice");
await invokeStart(bridge, functions, "alice");
const first = runtime.state.accepted[0];
assert.equal(store.turnOrigin(chatId, threadId, first).actorId, actors[bridge]);
runtime.state.pending = [{ id: "approval-a", turn_id: first }];
await store.patchChat(chatId, { authorizedIdentity: identities[bridge]("bob") });
runtime.state.failStart = true;
await assert.rejects(invokeStart(bridge, functions.runPrompt, "bob"), /start rejected/);
await assert.rejects(invokeStart(bridge, functions, "bob"), /start rejected/);
assert.equal(store.turnOrigin(chatId, threadId, first).actorId, actors[bridge]);
runtime.state.failStart = false;
runtime.state.turns = [{ id: first, status: "queued" }];
const before = runtime.state.startAttempts;
await invokeStart(bridge, functions.runPrompt, "bob");
await invokeStart(bridge, functions, "bob");
assert.equal(runtime.state.startAttempts, before);
runtime.state.turns = [];
await invokeStart(bridge, functions.runPrompt, "bob");
await invokeStart(bridge, functions, "bob");
assert.equal(store.turnOrigin(chatId, threadId, first).actorId, actors[bridge]);
const reopened = await ThreadStore.open(store.filePath, { actions: true });
const { decideApproval: consumer } = await shipping(bridge, ["turnActor", "decideApproval"], environment(reopened, runtime));
@@ -374,10 +395,11 @@ test("Telegram and Feishu shipping inbound events retain the actual admitted hum
assert.equal((await reopened.getChat(chat)).activeTurnId, turn, `${bridge} must persist the accepted turn before streaming`);
assert.equal(reopened.turnOrigin(chat, thread, turn).actorId, actors[bridge]);
delivered();
return true;
}
});
const names = bridge === "telegram-bridge" ?
["turnActor", "runPrompt", "startPromptTurn", "handleCommand", "rememberAuthorizedIdentity", "handleIncomingUpdate"] :
["turnActor", "runPrompt", "startPromptTurn", "startTrackedTurnStream", "handleCommand", "rememberAuthorizedIdentity", "handleIncomingUpdate"] :
["turnActor", "runPrompt", "handleCommand", "handleIncomingMessage"];
const handlers = await shipping(bridge, names, values);
if (bridge === "telegram-bridge") {
@@ -26,13 +26,13 @@ for(const platform of ['telegram','feishu']) {
await writer.setChat('chat',{threadId:'thread',activeTurnId:'turn',lastSeq:7,authorizedIdentity:identity,replyToMessageId:'original'});
const threadStore=await ThreadStore.open(file,{messageLimit:200});
const calls={runtime:[],sent:[],stream:[],commands:[]};
const context=vm.createContext({...lib,threadStore,config:{allowlist:['operator'],allowGroups:false,allowUnlisted:false,requirePrefixInGroup:false,groupPrefix:'/cw',...policy},
const context=vm.createContext({...lib,threadStore,botAccountId:'fixture-bot',config:{allowlist:['operator'],allowGroups:false,allowUnlisted:false,requirePrefixInGroup:false,groupPrefix:'/cw',...policy},
runtimeJson:async route=>{calls.runtime.push(route);return {turns:[{id:'turn',status:'in_progress'}]};},
sendText:async(...args)=>calls.sent.push(args),sendTurnText:async(...args)=>calls.sent.push(args),
streamTurnEvents:async(...args)=>calls.stream.push(args),startTrackedTurnStream:(...args)=>calls.stream.push(args),
handleCommand:async(...args)=>calls.commands.push(args),answerCallback:async()=>{},callbackAction:()=>({kind:'status'}),handleModalAction:async(...args)=>calls.commands.push(args),
});
const names=platform==='telegram'?['reattachActiveTurns','handleIncomingUpdate','handleCallbackQuery','rememberAuthorizedIdentity']:['reattachActiveTurns','handleIncomingMessage'];
const names=platform==='telegram'?['recoveryAllowed','reattachActiveTurns','handleIncomingUpdate','handleCallbackQuery','rememberAuthorizedIdentity']:['reattachActiveTurns','handleIncomingMessage'];
for(const name of names) {
if(name==='rememberAuthorizedIdentity' && !source.includes('function rememberAuthorizedIdentity(')) continue;
vm.runInContext(extract(source,name),context);
@@ -55,7 +55,7 @@ for(const platform of ['telegram','feishu']) {
for(const policy of [{allowlist:['operator']},{allowlist:['chat']},{allowUnlisted:true},{allowGroups:true}]) {
const identity=policy.allowGroups?{...baseIdentity,chatType:'group'}:baseIdentity;
const f=await fixture(t,identity,policy);await f.context.reattachActiveTurns();
assert.deepEqual(f.calls.runtime,['/v1/threads/thread']);assert.equal(f.calls.sent.length,1);assert.equal(f.calls.stream.length,1);
assert.deepEqual(f.calls.runtime,platform==='telegram'?[]:['/v1/threads/thread']);assert.equal(f.calls.sent.length,platform==='telegram'?0:1);assert.equal(f.calls.stream.length,1);
}
});
function incoming(userId,chatType=direct) {
+12
View File
@@ -0,0 +1,12 @@
DINGTALK_CLIENT_ID=
DINGTALK_CLIENT_SECRET=
DINGTALK_TENANT_ALLOWLIST=
DINGTALK_ACTOR_ALLOWLIST=
DINGTALK_ALLOW_GROUPS=false
DINGTALK_CONVERSATION_ALLOWLIST=
CODEWHALE_RUNTIME_URL=http://127.0.0.1:7331
CODEWHALE_RUNTIME_TOKEN=
CODEWHALE_MODEL=auto
CODEWHALE_WORKSPACE=
CODEWHALE_ALLOW_SHELL=false
DINGTALK_STATE_PATH=.state/dingtalk.json
+116
View File
@@ -0,0 +1,116 @@
# DingTalk bridge (advanced self-hosted integration)
This source adapter connects a DingTalk enterprise robot using Stream mode to an
existing `codewhale serve --http` Engine. It forwards tasks, approvals, answers,
stop requests and Engine events; the Engine owns prompts, tools, permissions,
turns and conversation history. No second agent runtime is started.
Codewhale-hosted bots are the intended default product experience. This optional
operator-managed adapter is for an enterprise's own DingTalk application and
Engine; its credential and Node.js setup is not the default user onboarding flow.
It does not implement hosted installation, Codewhale account linking, enterprise
ISV authorization or a tenant credential service.
The adapter is source-ready for offline verification. No DingTalk application,
credential, production connection or deployment is configured by this repository.
Live qualification requires a separately authorized tenant and runtime.
## Dependencies and setup
Use Node.js 22 or newer. From this directory, install the lockfile with
`npm ci --ignore-scripts`. The official MIT-licensed
[`dingtalk-stream`](https://github.com/open-dingtalk/dingtalk-stream-sdk-nodejs/tree/v2.1.7-beta.1)
SDK is pinned to registry release `2.1.7-beta.1`; this is an explicit prerelease
selection, including the SDK's reconnect and disconnect lifecycle fixes. The
lockfile pins its transitive dependencies. No SDK implementation was copied.
Create an enterprise robot with Stream mode and subscribe to robot messages.
The adapter uses the SDK's outgoing connection, not a public callback server.
Follow the official [Stream protocol](https://open-dingtalk.github.io/developerpedia/docs/learn/stream/protocol/)
and [SDK example](https://github.com/open-dingtalk/dingtalk-stream-sdk-nodejs/blob/v2.1.7-beta.1/example/index.ts).
Do not enable or connect an application until its owner has approved activation.
Copy `.env.example` to `.env`, fill it locally, and use
`node --env-file=.env src/index.mjs --validate-config` to validate without opening
any connection. The normal entry point is `node --env-file=.env src/index.mjs`.
`npm run check` performs syntax checks and does not connect.
| Setting | Meaning |
| --- | --- |
| `DINGTALK_CLIENT_ID`, `DINGTALK_CLIENT_SECRET` | Enterprise application credentials |
| `DINGTALK_TENANT_ALLOWLIST` | Comma-separated exact corporate IDs; both robot and sender corporation must be listed |
| `DINGTALK_ACTOR_ALLOWLIST` | Comma-separated exact `senderCorpId:senderStaffId` pairs |
| `DINGTALK_ALLOW_GROUPS` | `false` by default; only literal `true` or `false` is accepted |
| `DINGTALK_CONVERSATION_ALLOWLIST` | Required for groups; exact `chatbotCorpId:conversationId` pairs |
| `CODEWHALE_RUNTIME_URL`, `CODEWHALE_RUNTIME_TOKEN` | Existing authenticated Engine HTTP endpoint; HTTPS or loopback HTTP |
| `CODEWHALE_MODEL` | Existing Engine model or route, defaults to `auto` |
| `CODEWHALE_WORKSPACE` | Optional workspace passed to Engine thread creation |
| `CODEWHALE_ALLOW_SHELL` | Defaults to `false`; enabling shell availability still requires Engine approval |
| `DINGTALK_STATE_PATH` | Defaults to `.state/dingtalk.json` relative to the working directory |
Tenant and actor allowlists are mandatory; there is no allow-unlisted option.
Group requests also require `isInAtList: true` on every message. Sessions are
separate for each application, bot tenant, conversation, sender tenant and staff
ID. Group responses are visible to that group, so only enable appropriate groups.
Allowlisting a user does not enable trust mode or blanket approval: both are
always disabled on this adapter's requests.
## Conversation controls
Send a text task, then use these commands in the same conversation and identity:
- `/status` reports the bound thread, active turn, retained admission and uncertain deliveries.
- `/stop` or `/interrupt` requests cancellation of that human's accepted active turn.
- `/allow APPROVAL_ID` or `/deny APPROVAL_ID` decides one live Engine approval.
- `/answer QUESTION_ID response` answers a live question. Multiple questions need one answer per line. A plain reply also answers the sole pending question for the current turn.
- `/help` shows the available controls.
Every approval and answer is checked against the current Engine pending request
and immutable recorded initiating actor. Another user's message cannot authorize
the turn. Incoming `sessionWebhook` credentials never enter the Engine prompt.
## Durability and recovery
Run one process per state file. An exclusive `.lock` file rejects concurrent
instances. State is private (`0700` directory and `0600` file), replaced through
the shared fsync-backed store. Retain it across restarts; it contains sensitive
conversation text and short-lived reply credentials. Do not commit or publish it.
The adapter durably saves an authenticated inbox entry and its message identity
before acknowledging Stream delivery. It retains 2,000 completed message keys;
queued messages and pending admissions are independently retained. The queue is
bounded, and a full queue is not acknowledged. Unsupported or unauthorized input
is acknowledged without contacting the Engine. Text-only input is supported.
Turn submission uses the shared bridge-core admission helper. When the Engine
advertises both operation idempotency and lookup, the original request and key
are retained, and recovery queries that exact key before retrying. Only an exact
404 permits the unchanged request to be submitted again. Conflicts and lookup
failures retain the receipt. Without both capabilities, an uncertain submission
is never repeated. Reconnect resumes the same Engine event cursor and accepted
turn. SDK Stream reconnect is enabled independently of Engine SSE reconnect.
Each output chunk is saved before sending and marked in flight before its one
HTTP attempt. A timeout, rejected response, restart during sending or uncertain
receipt marks that chunk uncertain; it is not resent automatically. A successful
response requires HTTP success and DingTalk `errcode: 0`. Unattempted output waits
when its session webhook expires, and a new authorized message can supply a fresh
route for the same session. Only HTTPS `oapi.dingtalk.com/robot/sendBySession`
routes with a future expiry are accepted; redirects are rejected. No proactive
message API or alternate recipient is used. SDK warning objects are suppressed
in the dedicated process to avoid exposing transient connection credentials.
After an abrupt shutdown, verify that no bridge process still owns the state
file before removing a stale `.lock`. Back up the state first. Operators must
reconcile uncertain thread creation, non-idempotent command effects, admission
without lookup support, and uncertain output against Engine/DingTalk receipts.
Do not remove retained receipts or reset the state as a retry mechanism. A
thread-creation request without a confirmed ID is deliberately not retried.
This first adapter does not support attachments, interactive cards, sharing a
single conversation between different staff identities, browsing existing Engine
threads or automatic output reconciliation. Commands interrupted during a crash
are reported uncertain rather than replayed. Response capture is capped at one
million characters with an explicit transcript notice. A host filesystem and one
process provide durability; this is not a distributed queue. Keep provider calls,
tenant messaging and production deployment separate from offline checks.
+381
View File
@@ -0,0 +1,381 @@
{
"name": "@codewhale/dingtalk-bridge",
"version": "0.1.0",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "@codewhale/dingtalk-bridge",
"version": "0.1.0",
"dependencies": {
"dingtalk-stream": "2.1.7-beta.1"
},
"engines": {
"node": ">=22"
}
},
"node_modules/agent-base": {
"version": "6.0.2",
"resolved": "https://registry.npmjs.org/agent-base/-/agent-base-6.0.2.tgz",
"integrity": "sha512-RZNwNclF7+MS/8bDg70amg32dyeZGZxiDuQmZxKLAlQjr3jGyLx+4Kkk58UO7D2QdgFIQCovuSuZESne6RG6XQ==",
"license": "MIT",
"dependencies": {
"debug": "4"
},
"engines": {
"node": ">= 6.0.0"
}
},
"node_modules/asynckit": {
"version": "0.4.0",
"resolved": "https://registry.npmjs.org/asynckit/-/asynckit-0.4.0.tgz",
"integrity": "sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q==",
"license": "MIT"
},
"node_modules/axios": {
"version": "1.20.0",
"resolved": "https://registry.npmjs.org/axios/-/axios-1.20.0.tgz",
"integrity": "sha512-r8aOh8j9cGKpgQAqpzrUHnSIc6a59Y3Xf/cv8sy1DrHCkZHzQGEuoq1tARk6qSyDdtQGSDgpb9kFlruzPvrgwg==",
"license": "MIT",
"dependencies": {
"follow-redirects": "^1.16.0",
"form-data": "^4.0.6",
"https-proxy-agent": "^5.0.1",
"proxy-from-env": "^2.1.0"
}
},
"node_modules/call-bind-apply-helpers": {
"version": "1.0.2",
"resolved": "https://registry.npmjs.org/call-bind-apply-helpers/-/call-bind-apply-helpers-1.0.2.tgz",
"integrity": "sha512-Sp1ablJ0ivDkSzjcaJdxEunN5/XvksFJ2sMBFfq6x0ryhQV/2b/KwFe21cMpmHtPOSij8K99/wSfoEuTObmuMQ==",
"license": "MIT",
"dependencies": {
"es-errors": "^1.3.0",
"function-bind": "^1.1.2"
},
"engines": {
"node": ">= 0.4"
}
},
"node_modules/combined-stream": {
"version": "1.0.8",
"resolved": "https://registry.npmjs.org/combined-stream/-/combined-stream-1.0.8.tgz",
"integrity": "sha512-FQN4MRfuJeHf7cBbBMJFXhKSDq+2kAArBlmRBvcvFE5BB1HZKXtSFASDhdlz9zOYwxh8lDdnvmMOe/+5cdoEdg==",
"license": "MIT",
"dependencies": {
"delayed-stream": "~1.0.0"
},
"engines": {
"node": ">= 0.8"
}
},
"node_modules/debug": {
"version": "4.4.3",
"resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz",
"integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==",
"license": "MIT",
"dependencies": {
"ms": "^2.1.3"
},
"engines": {
"node": ">=6.0"
},
"peerDependenciesMeta": {
"supports-color": {
"optional": true
}
}
},
"node_modules/delayed-stream": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/delayed-stream/-/delayed-stream-1.0.0.tgz",
"integrity": "sha512-ZySD7Nf91aLB0RxL4KGrKHBXl7Eds1DAmEdcoVawXnLD7SDhpNgtuII2aAkg7a7QS41jxPSZ17p4VdGnMHk3MQ==",
"license": "MIT",
"engines": {
"node": ">=0.4.0"
}
},
"node_modules/dingtalk-stream": {
"version": "2.1.7-beta.1",
"resolved": "https://registry.npmjs.org/dingtalk-stream/-/dingtalk-stream-2.1.7-beta.1.tgz",
"integrity": "sha512-MBWfDa3KrocwFPf5tOAhG2oM6FFYBN30sQQ7wbz7BPlRcuemAb7SGXbeztS49Pp/3Hiyzhwqjcssf8Yq3Z9DzA==",
"license": "MIT",
"dependencies": {
"axios": "^1.19.0",
"debug": "^4.3.4",
"ws": "^8.21.1"
}
},
"node_modules/dunder-proto": {
"version": "1.0.1",
"resolved": "https://registry.npmjs.org/dunder-proto/-/dunder-proto-1.0.1.tgz",
"integrity": "sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==",
"license": "MIT",
"dependencies": {
"call-bind-apply-helpers": "^1.0.1",
"es-errors": "^1.3.0",
"gopd": "^1.2.0"
},
"engines": {
"node": ">= 0.4"
}
},
"node_modules/es-define-property": {
"version": "1.0.1",
"resolved": "https://registry.npmjs.org/es-define-property/-/es-define-property-1.0.1.tgz",
"integrity": "sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g==",
"license": "MIT",
"engines": {
"node": ">= 0.4"
}
},
"node_modules/es-errors": {
"version": "1.3.0",
"resolved": "https://registry.npmjs.org/es-errors/-/es-errors-1.3.0.tgz",
"integrity": "sha512-Zf5H2Kxt2xjTvbJvP2ZWLEICxA6j+hAmMzIlypy4xcBg1vKVnx89Wy0GbS+kf5cwCVFFzdCFh2XSCFNULS6csw==",
"license": "MIT",
"engines": {
"node": ">= 0.4"
}
},
"node_modules/es-object-atoms": {
"version": "1.1.2",
"resolved": "https://registry.npmjs.org/es-object-atoms/-/es-object-atoms-1.1.2.tgz",
"integrity": "sha512-HWcBoN6NileqtSydK2FqHbS/LoDd2pqrnQHLyJzBj4kOp/ky2MWMN694xOfkK8/SnUsW2DH7EfyVlydKCsm1Zw==",
"license": "MIT",
"dependencies": {
"es-errors": "^1.3.0"
},
"engines": {
"node": ">= 0.4"
}
},
"node_modules/es-set-tostringtag": {
"version": "2.1.0",
"resolved": "https://registry.npmjs.org/es-set-tostringtag/-/es-set-tostringtag-2.1.0.tgz",
"integrity": "sha512-j6vWzfrGVfyXxge+O0x5sh6cvxAog0a/4Rdd2K36zCMV5eJ+/+tOAngRO8cODMNWbVRdVlmGZQL2YS3yR8bIUA==",
"license": "MIT",
"dependencies": {
"es-errors": "^1.3.0",
"get-intrinsic": "^1.2.6",
"has-tostringtag": "^1.0.2",
"hasown": "^2.0.2"
},
"engines": {
"node": ">= 0.4"
}
},
"node_modules/follow-redirects": {
"version": "1.16.2",
"resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.16.2.tgz",
"integrity": "sha512-/kVP2U8z596Ia0gwFrwyIeo74ta5aVmSeU1FfAUMAmcEP5EP7zydmag3SLiN64NaRKef3lxsq7i/g7ltJQDGow==",
"funding": [
{
"type": "individual",
"url": "https://github.com/sponsors/RubenVerborgh"
}
],
"license": "MIT",
"engines": {
"node": ">=4.0"
},
"peerDependenciesMeta": {
"debug": {
"optional": true
}
}
},
"node_modules/form-data": {
"version": "4.0.6",
"resolved": "https://registry.npmjs.org/form-data/-/form-data-4.0.6.tgz",
"integrity": "sha512-vKatAh4SlVfgbv+YtmhiRjhEMJsYpsG1Y2rMQtR+SVSbytsSD1YGzDIcrAJmdFec88u/+VoGmxnl+80gL1tRCQ==",
"license": "MIT",
"dependencies": {
"asynckit": "^0.4.0",
"combined-stream": "^1.0.8",
"es-set-tostringtag": "^2.1.0",
"hasown": "^2.0.4",
"mime-types": "^2.1.35"
},
"engines": {
"node": ">= 6"
}
},
"node_modules/function-bind": {
"version": "1.1.2",
"resolved": "https://registry.npmjs.org/function-bind/-/function-bind-1.1.2.tgz",
"integrity": "sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA==",
"license": "MIT",
"funding": {
"url": "https://github.com/sponsors/ljharb"
}
},
"node_modules/get-intrinsic": {
"version": "1.3.0",
"resolved": "https://registry.npmjs.org/get-intrinsic/-/get-intrinsic-1.3.0.tgz",
"integrity": "sha512-9fSjSaos/fRIVIp+xSJlE6lfwhES7LNtKaCBIamHsjr2na1BiABJPo0mOjjz8GJDURarmCPGqaiVg5mfjb98CQ==",
"license": "MIT",
"dependencies": {
"call-bind-apply-helpers": "^1.0.2",
"es-define-property": "^1.0.1",
"es-errors": "^1.3.0",
"es-object-atoms": "^1.1.1",
"function-bind": "^1.1.2",
"get-proto": "^1.0.1",
"gopd": "^1.2.0",
"has-symbols": "^1.1.0",
"hasown": "^2.0.2",
"math-intrinsics": "^1.1.0"
},
"engines": {
"node": ">= 0.4"
},
"funding": {
"url": "https://github.com/sponsors/ljharb"
}
},
"node_modules/get-proto": {
"version": "1.0.1",
"resolved": "https://registry.npmjs.org/get-proto/-/get-proto-1.0.1.tgz",
"integrity": "sha512-sTSfBjoXBp89JvIKIefqw7U2CCebsc74kiY6awiGogKtoSGbgjYE/G/+l9sF3MWFPNc9IcoOC4ODfKHfxFmp0g==",
"license": "MIT",
"dependencies": {
"dunder-proto": "^1.0.1",
"es-object-atoms": "^1.0.0"
},
"engines": {
"node": ">= 0.4"
}
},
"node_modules/gopd": {
"version": "1.2.0",
"resolved": "https://registry.npmjs.org/gopd/-/gopd-1.2.0.tgz",
"integrity": "sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg==",
"license": "MIT",
"engines": {
"node": ">= 0.4"
},
"funding": {
"url": "https://github.com/sponsors/ljharb"
}
},
"node_modules/has-symbols": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/has-symbols/-/has-symbols-1.1.0.tgz",
"integrity": "sha512-1cDNdwJ2Jaohmb3sg4OmKaMBwuC48sYni5HUw2DvsC8LjGTLK9h+eb1X6RyuOHe4hT0ULCW68iomhjUoKUqlPQ==",
"license": "MIT",
"engines": {
"node": ">= 0.4"
},
"funding": {
"url": "https://github.com/sponsors/ljharb"
}
},
"node_modules/has-tostringtag": {
"version": "1.0.2",
"resolved": "https://registry.npmjs.org/has-tostringtag/-/has-tostringtag-1.0.2.tgz",
"integrity": "sha512-NqADB8VjPFLM2V0VvHUewwwsw0ZWBaIdgo+ieHtK3hasLz4qeCRjYcqfB6AQrBggRKppKF8L52/VqdVsO47Dlw==",
"license": "MIT",
"dependencies": {
"has-symbols": "^1.0.3"
},
"engines": {
"node": ">= 0.4"
},
"funding": {
"url": "https://github.com/sponsors/ljharb"
}
},
"node_modules/hasown": {
"version": "2.0.4",
"resolved": "https://registry.npmjs.org/hasown/-/hasown-2.0.4.tgz",
"integrity": "sha512-T2UbfbBEF32wiepXIsMlTW9+dDYC6wMh/t/vYA4tuOMKqWz/n3vr1NFSxQiyP+zk2mXsoMA/i/7qV6LKut1t1A==",
"license": "MIT",
"dependencies": {
"function-bind": "^1.1.2"
},
"engines": {
"node": ">= 0.4"
}
},
"node_modules/https-proxy-agent": {
"version": "5.0.1",
"resolved": "https://registry.npmjs.org/https-proxy-agent/-/https-proxy-agent-5.0.1.tgz",
"integrity": "sha512-dFcAjpTQFgoLMzC2VwU+C/CbS7uRL0lWmxDITmqm7C+7F0Odmj6s9l6alZc6AELXhrnggM2CeWSXHGOdX2YtwA==",
"license": "MIT",
"dependencies": {
"agent-base": "6",
"debug": "4"
},
"engines": {
"node": ">= 6"
}
},
"node_modules/math-intrinsics": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/math-intrinsics/-/math-intrinsics-1.1.0.tgz",
"integrity": "sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g==",
"license": "MIT",
"engines": {
"node": ">= 0.4"
}
},
"node_modules/mime-db": {
"version": "1.52.0",
"resolved": "https://registry.npmjs.org/mime-db/-/mime-db-1.52.0.tgz",
"integrity": "sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg==",
"license": "MIT",
"engines": {
"node": ">= 0.6"
}
},
"node_modules/mime-types": {
"version": "2.1.35",
"resolved": "https://registry.npmjs.org/mime-types/-/mime-types-2.1.35.tgz",
"integrity": "sha512-ZDY+bPm5zTTF+YpCrAU9nK0UgICYPT0QtT1NZWFv4s++TNkcgVaT0g6+4R2uI4MjQjzysHB1zxuWL50hzaeXiw==",
"license": "MIT",
"dependencies": {
"mime-db": "1.52.0"
},
"engines": {
"node": ">= 0.6"
}
},
"node_modules/ms": {
"version": "2.1.3",
"resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz",
"integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==",
"license": "MIT"
},
"node_modules/proxy-from-env": {
"version": "2.1.0",
"resolved": "https://registry.npmjs.org/proxy-from-env/-/proxy-from-env-2.1.0.tgz",
"integrity": "sha512-cJ+oHTW1VAEa8cJslgmUZrc+sjRKgAKl3Zyse6+PV38hZe/V6Z14TbCuXcan9F9ghlz4QrFr2c92TNF82UkYHA==",
"license": "MIT",
"engines": {
"node": ">=10"
}
},
"node_modules/ws": {
"version": "8.22.0",
"resolved": "https://registry.npmjs.org/ws/-/ws-8.22.0.tgz",
"integrity": "sha512-Ydggc987+RO0AnWtZ/7Wq9FtNvcrL1b/RO0ud9mWjUPgDrsAAwQSF51sm2hm1XofbU/4jkpGEsLFsZZxU+1DOg==",
"license": "MIT",
"engines": {
"node": ">=10.0.0"
},
"peerDependencies": {
"bufferutil": "^4.0.1",
"utf-8-validate": ">=5.0.2"
},
"peerDependenciesMeta": {
"bufferutil": {
"optional": true
},
"utf-8-validate": {
"optional": true
}
}
}
}
}
+14
View File
@@ -0,0 +1,14 @@
{
"name": "@codewhale/dingtalk-bridge",
"version": "0.1.0",
"private": true,
"type": "module",
"description": "DingTalk Stream transport for the Codewhale Engine HTTP API.",
"scripts": {
"start": "node src/index.mjs",
"check": "node --check src/index.mjs && node --check src/lib.mjs",
"validate:config": "node src/index.mjs --validate-config"
},
"engines": { "node": ">=22" },
"dependencies": { "dingtalk-stream": "2.1.7-beta.1" }
}
@@ -0,0 +1,62 @@
import { ThreadStore, createRuntimeClient } from "../../bridge-core/src/lib.mjs";
import { DingTalkBridge, createReplySender, readConfig } from "./lib.mjs";
import { mkdir, open, unlink } from "node:fs/promises";
import path from "node:path";
async function main() {
const config = readConfig();
if (process.argv.includes("--validate-config")) {
console.log("DingTalk bridge configuration is valid; no connection was opened.");
return;
}
const sdkWarning = console.warn.bind(console);
console.warn = () => sdkWarning("DingTalk SDK warning; connection state is retained.");
const { DWClient, TOPIC_ROBOT } = await import("dingtalk-stream");
const client = new DWClient({ clientId: config.clientId, clientSecret: config.clientSecret, debug: false, autoReconnect: true });
await mkdir(path.dirname(config.statePath), { recursive: true, mode: 0o700 });
const lockPath = `${config.statePath}.lock`;
const lock = await open(lockPath, "wx", 0o600);
let store;
try {
await lock.writeFile(String(process.pid));
await lock.sync();
store = await ThreadStore.open(config.statePath, { messageLimit: 2000, privateMode: true });
} catch (error) {
await lock.close();
await unlink(lockPath);
throw error;
}
const { runtimeJson, authHeaders } = createRuntimeClient(config);
const bridge = new DingTalkBridge({
config, store, runtimeJson, sendReply: createReplySender({ client }),
runtimeFetch: (route, signal) => fetch(`${config.runtimeUrl}${route}`, { headers: { ...authHeaders(), accept: "text/event-stream" }, signal: AbortSignal.any([signal, AbortSignal.timeout(60000)]) }),
report: (message) => console.error(message)
});
client.registerCallbackListener(TOPIC_ROBOT, async (frame) => {
try {
await bridge.accept(frame, () => client.socketCallBackResponse(frame.headers.messageId, { errcode: 0, errmsg: "ok" }));
} catch {
console.error("DingTalk message was not acknowledged; durable acceptance failed.");
}
});
let closing = false;
const shutdown = async () => {
if (closing) return;
closing = true;
client.disconnect();
await bridge.stop();
await lock.close();
await unlink(lockPath);
};
for (const signal of ["SIGINT", "SIGTERM"]) process.once(signal, () => { void shutdown().catch(() => { process.exitCode = 1; }); });
try {
await bridge.start();
if (!closing) await client.connect();
if (!closing) console.log("DingTalk Stream bridge started; SDK reconnect is enabled.");
} catch (error) {
await shutdown();
throw error;
}
}
main().catch(() => { console.error("DingTalk bridge startup failed. Check configuration, runtime availability and the state lock."); process.exitCode = 1; });
+366
View File
@@ -0,0 +1,366 @@
import { createHash } from "node:crypto";
import path from "node:path";
import { setTimeout as delay } from "node:timers/promises";
import {
admitTurn, decideApproval, parseQuestionAnswers, questionText, readSse, splitMessage
} from "../../bridge-core/src/lib.mjs";
const digest = (value) => createHash("sha256").update(JSON.stringify(value)).digest("hex");
const list = (value) => new Set(String(value || "").split(/[\s,]+/u).filter(Boolean));
const bool = (value, name) => {
if (value === undefined || value === "false") return false;
if (value === "true") return true;
throw new Error(`${name} must be true or false`);
};
const required = (env, name) => {
if (!env[name]?.trim()) throw new Error(`${name} is required`);
return env[name].trim();
};
export function readConfig(env = process.env) {
const runtime = new URL(required(env, "CODEWHALE_RUNTIME_URL"));
if (runtime.username || runtime.password || runtime.search || runtime.hash ||
(runtime.protocol !== "https:" && !(runtime.protocol === "http:" && ["localhost", "127.0.0.1", "[::1]"].includes(runtime.hostname)))) {
throw new Error("Runtime URL requires HTTPS or loopback HTTP without credentials, query or fragment");
}
const config = {
clientId: required(env, "DINGTALK_CLIENT_ID"), clientSecret: required(env, "DINGTALK_CLIENT_SECRET"),
tenants: list(required(env, "DINGTALK_TENANT_ALLOWLIST")), actors: list(required(env, "DINGTALK_ACTOR_ALLOWLIST")),
conversations: list(env.DINGTALK_CONVERSATION_ALLOWLIST), allowGroups: bool(env.DINGTALK_ALLOW_GROUPS, "DINGTALK_ALLOW_GROUPS"),
runtimeUrl: runtime.href.replace(/\/$/u, ""), runtimeToken: required(env, "CODEWHALE_RUNTIME_TOKEN"),
model: env.CODEWHALE_MODEL?.trim() || "auto", workspace: env.CODEWHALE_WORKSPACE?.trim() || null,
allowShell: bool(env.CODEWHALE_ALLOW_SHELL, "CODEWHALE_ALLOW_SHELL"),
statePath: path.resolve(env.DINGTALK_STATE_PATH || ".state/dingtalk.json"), runtimeRequestTimeoutMs: 15000
};
if (!config.tenants.size || !config.actors.size || (config.allowGroups && !config.conversations.size)) {
throw new Error("DingTalk requires nonempty tenant and actor allowlists, and an explicit conversation allowlist for groups");
}
return config;
}
export function authorized(config, identity) {
return Boolean(identity && config.tenants.has(identity.botTenant) && config.tenants.has(identity.senderTenant) &&
config.actors.has(`${identity.senderTenant}:${identity.staffId}`) &&
(identity.kind === "1" || (identity.kind === "2" && config.allowGroups && config.conversations.has(`${identity.botTenant}:${identity.conversationId}`))));
}
export function replyRoute(data, now = Date.now()) {
const url = new URL(data.sessionWebhook);
if (url.protocol !== "https:" || url.hostname !== "oapi.dingtalk.com" || url.port || url.username || url.password || url.hash ||
url.pathname !== "/robot/sendBySession" || !url.search || !Number.isSafeInteger(data.sessionWebhookExpiredTime) || data.sessionWebhookExpiredTime <= now) {
throw new Error("DingTalk reply route is invalid or expired");
}
return { url: url.href, expiresAt: data.sessionWebhookExpiredTime };
}
export function incoming(config, frame, now = Date.now()) {
if (typeof frame?.headers?.messageId !== "string" || typeof frame.data !== "string" || frame.data.length > 65536) return null;
const data = JSON.parse(frame.data);
const identity = {
botTenant: data.chatbotCorpId, senderTenant: data.senderCorpId, staffId: data.senderStaffId,
conversationId: data.conversationId, kind: data.conversationType
};
if (Object.values(identity).some((value) => typeof value !== "string" || !value || value.length > 512) || !authorized(config, identity) ||
(identity.kind === "2" && data.isInAtList !== true) || data.msgtype !== "text" || typeof data.msgId !== "string" || !data.msgId ||
typeof data.text?.content !== "string" || !data.text.content.trim() || data.text.content.length > 20000) return null;
const tuple = [config.clientId, identity.botTenant, identity.conversationId, identity.senderTenant, identity.staffId];
return {
chatId: digest(tuple), actorId: `dingtalk:${identity.senderTenant}:${identity.staffId}`, identity,
messageKey: digest([...tuple, data.msgId]), text: data.text.content.trim(), route: replyRoute(data, now)
};
}
export function createReplySender({ client, fetchImpl = fetch, now = Date.now }) {
return async (route, content) => {
replyRoute({ sessionWebhook: route.url, sessionWebhookExpiredTime: route.expiresAt }, now());
const token = await client.getAccessToken();
replyRoute({ sessionWebhook: route.url, sessionWebhookExpiredTime: route.expiresAt }, now());
const response = await fetchImpl(route.url, {
method: "POST", redirect: "error", signal: AbortSignal.timeout(15000),
headers: { "content-type": "application/json", "x-acs-dingtalk-access-token": token },
body: JSON.stringify({ msgtype: "text", text: { content } })
});
const result = await response.json();
if (!response.ok || result?.errcode !== 0) throw new Error("DingTalk delivery was not confirmed");
};
}
export class DingTalkBridge {
constructor({ config, store, runtimeJson, runtimeFetch, sendReply, now = Date.now, report = () => {} }) {
Object.assign(this, { config, store, runtimeJson, runtimeFetch, sendReply, now, report });
this.serial = Promise.resolve();
this.tasks = new Map();
this.again = new Map();
this.closed = false;
this.abort = new AbortController();
this.capabilities = {};
}
mutate(operation) {
const result = this.serial.then(operation);
this.serial = result.catch(() => {});
return result;
}
valid(state) {
return state?.clientId === this.config.clientId && state.runtimeUrl === this.config.runtimeUrl && authorized(this.config, state.identity) &&
state.actorId === `dingtalk:${state.identity.senderTenant}:${state.identity.staffId}`;
}
output(state, id, text) {
const outbox = state.outbox || [];
if (outbox.some((entry) => entry.id === id)) return outbox;
return [...outbox.filter((entry) => entry.status !== "sent").concat(outbox.filter((entry) => entry.status === "sent").slice(-100)),
{ id, chunks: splitMessage(text || "Turn finished without a text response.", 2000), index: 0, status: "pending" }];
}
async start() {
this.capabilities = (await this.runtimeJson("/v1/runtime/info"))?.capabilities || {};
for (const [chatId] of this.store.listChats()) {
await this.mutate(async () => {
const state = await this.store.getChat(chatId);
if (!this.valid(state)) return;
let outbox = (state.outbox || []).map((entry) => entry.status === "sending" ? { ...entry, status: "uncertain" } : entry);
const interrupted = (state.inbox || []).filter((entry) => entry.processing);
for (const entry of interrupted) outbox = this.output({ outbox }, `${entry.messageKey}:uncertain`, "A previous command may have reached the Engine. Its result is uncertain; it was not repeated. Use /status to inspect the current turn.");
await this.store.patchChat(chatId, { outbox, inbox: (state.inbox || []).filter((entry) => !entry.processing) });
});
this.kick(chatId);
}
}
async accept(frame, acknowledge) {
let message;
try { message = incoming(this.config, frame, this.now()); } catch { message = null; }
if (!message) { acknowledge(); return false; }
await this.mutate(async () => {
const state = await this.store.getChat(message.chatId) || {};
if (state.clientId && !this.valid(state)) throw new Error("DingTalk binding no longer matches current policy");
const inbox = state.inbox || [];
if (inbox.some((entry) => entry.messageKey === message.messageKey) || state.pendingAdmission?.messageKey === message.messageKey) { await this.store.save(); return; }
if (inbox.length >= 100 || (state.outbox || []).filter((entry) => entry.status !== "sent").length >= 100) throw new Error("DingTalk durable queue is full");
await this.store.recordMessage(message.messageKey, message.chatId, {
clientId: this.config.clientId, runtimeUrl: this.config.runtimeUrl, identity: message.identity, actorId: message.actorId,
route: message.route, inbox: [...inbox, { messageKey: message.messageKey, text: message.text }]
});
await this.store.save();
});
acknowledge();
this.kick(message.chatId);
return true;
}
spawn(key, operation) {
if (this.closed) return;
if (this.tasks.has(key)) { this.again.set(key, operation); return; }
const task = Promise.resolve().then(operation).catch(() => this.report("DingTalk transport work retained for recovery"))
.finally(() => {
this.tasks.delete(key);
const next = this.again.get(key);
this.again.delete(key);
if (next) this.spawn(key, next);
});
this.tasks.set(key, task);
}
kick(chatId) {
this.spawn(`inbox:${chatId}`, () => this.drain(chatId));
this.spawn(`turn:${chatId}`, () => this.runTurn(chatId));
this.spawn(`output:${chatId}`, () => this.flush(chatId));
}
async enqueue(chatId, id, text) {
await this.mutate(async () => {
const state = await this.store.getChat(chatId);
await this.store.patchChat(chatId, { outbox: this.output(state, id, text) });
});
this.spawn(`output:${chatId}`, () => this.flush(chatId));
}
async drain(chatId) {
for (;;) {
const entry = await this.mutate(async () => {
const state = await this.store.getChat(chatId);
if (!this.valid(state) || !state.inbox?.length) return null;
const entry = { ...state.inbox[0], processing: true };
await this.store.patchChat(chatId, { inbox: [entry, ...state.inbox.slice(1)] });
return entry;
});
if (!entry || this.closed) return;
try {
await this.handle(chatId, entry);
} catch {
await this.enqueue(chatId, `${entry.messageKey}:error`, "The request could not be confirmed. It will not be repeated automatically. Use /status to inspect the retained state.");
}
await this.mutate(async () => {
const state = await this.store.getChat(chatId);
await this.store.patchChat(chatId, { inbox: state.inbox.filter((item) => item.messageKey !== entry.messageKey) });
});
this.spawn(`turn:${chatId}`, () => this.runTurn(chatId));
}
}
async handle(chatId, entry) {
const state = await this.store.getChat(chatId);
const match = /^\/(\w+)(?:\s+([\s\S]*))?$/u.exec(entry.text);
const command = match?.[1]?.toLowerCase();
const argument = match?.[2]?.trim() || "";
const reply = (text) => this.enqueue(chatId, `${entry.messageKey}:reply`, text);
if (command === "help") return reply("Send a task, /status, /stop, /allow APPROVAL_ID, /deny APPROVAL_ID, or /answer QUESTION_ID followed by your answers. Group conversations require an @mention. Only the initiating human can act on a turn.");
if (command === "status") return reply(`Thread: ${state.threadId || "not created"}\nTurn: ${state.active?.turnId || "none"}\nAdmission: ${state.pendingAdmission ? "retained; reconciliation pending" : state.threadCreateSubmitted && !state.threadId ? "thread creation uncertain" : "none"}\nUncertain deliveries: ${(state.outbox || []).filter((item) => item.status === "uncertain").length}`);
if (["allow", "deny"].includes(command)) {
if (!argument || /\s/u.test(argument)) return reply("Specify one approval ID.");
await decideApproval(this.runtimeJson, { store: this.store, chatId, actorId: state.actorId, approvalId: argument, decision: command === "allow" ? "allow" : "deny", remember: false });
return reply("Approval decision recorded.");
}
if (["stop", "interrupt"].includes(command)) {
if (!state.active || this.store.turnOrigin(chatId, state.threadId, state.active.turnId)?.actorId !== state.actorId) return reply("No owned active turn to stop.");
await this.runtimeJson(`/v1/threads/${encodeURIComponent(state.threadId)}/turns/${encodeURIComponent(state.active.turnId)}/interrupt`, { method: "POST" });
return reply("Stop requested.");
}
if (command === "answer" || (!command && state.active)) {
const detail = await this.runtimeJson(`/v1/threads/${encodeURIComponent(state.threadId)}`);
const inputId = argument.split(/\s/u)[0];
const inputs = (detail.pending_user_inputs || []).filter((input) => input.turn_id === state.active?.turnId &&
this.store.turnOrigin(chatId, state.threadId, input.turn_id)?.actorId === state.actorId);
const input = command === "answer" ? inputs.find((item) => (item.id || item.input_id) === inputId) : inputs.length === 1 ? inputs[0] : null;
if (input) {
const answerText = command === "answer" ? argument.slice(inputId.length).trim() : entry.text;
const answers = parseQuestionAnswers(input.request, answerText);
if (!answers) return reply(questionText(input.request));
await this.runtimeJson(`/v1/user-input/${encodeURIComponent(state.threadId)}/${encodeURIComponent(input.id || input.input_id)}`, { method: "POST", body: { answers } });
return reply("Answer recorded.");
}
if (command === "answer") return reply("That question is not pending for your current turn.");
}
if (command) return reply("Unknown command. Use /help.");
if (state.active || state.pendingAdmission) return reply("Your previous task is still active or awaiting reconciliation. Use /status or /stop.");
const operationKey = this.capabilities.turn_operation_idempotency === true && this.capabilities.turn_operation_lookup === true ? `dingtalk-${entry.messageKey}` : null;
const settings = { model: this.config.model, mode: "agent", allow_shell: this.config.allowShell, trust_mode: false, auto_approve: false };
const pending = {
messageKey: entry.messageKey, actorId: state.actorId, operationKey, threadId: state.threadId || null, submitted: false, sinceSeq: null,
threadRequest: { ...settings, workspace: this.config.workspace, archived: false },
request: { ...settings, prompt: entry.text, input_summary: entry.text, ...(operationKey ? { operation_key: operationKey } : {}) }
};
await this.mutate(async () => {
const current = await this.store.getChat(chatId);
await this.store.patchChat(chatId, { pendingAdmission: pending, inbox: current.inbox.filter((item) => item.messageKey !== entry.messageKey) });
});
}
async runTurn(chatId) {
while (!this.closed) {
let state = await this.store.getChat(chatId);
if (!this.valid(state)) return;
try {
if (state.pendingAdmission) {
const admission = await admitTurn({
runtimeJson: this.runtimeJson, pending: state.pendingAdmission, capabilities: this.capabilities,
persistPending: (pendingAdmission) => this.store.patchChat(chatId, { pendingAdmission }),
ensureThread: async (body) => {
const current = await this.store.getChat(chatId);
if (current.threadId) return current.threadId;
if (current.threadCreateSubmitted) throw new Error("Thread creation is uncertain");
await this.store.patchChat(chatId, { threadCreateSubmitted: true });
const thread = await this.runtimeJson("/v1/threads", { method: "POST", body });
if (typeof thread?.id !== "string" || !thread.id) throw new Error("Invalid Engine thread receipt");
await this.store.patchChat(chatId, { threadId: thread.id });
return thread.id;
}
});
if (!admission.turn) {
await this.enqueue(chatId, `${state.pendingAdmission.messageKey}:admission`, admission.uncertain ? "Task admission is uncertain and cannot be retried safely. An operator must reconcile the Engine receipt." : "Another turn is running in this Engine thread. Your task is retained until it finishes.");
if (admission.uncertain) return;
await delay(3000, undefined, { signal: this.abort.signal });
continue;
}
await this.store.recordTurnOrigin(chatId, admission.pending.threadId, admission.turn.id, admission.pending.actorId);
await this.store.patchChat(chatId, { pendingAdmission: null, active: { turnId: admission.turn.id, seq: admission.pending.sinceSeq, text: "" } });
state = await this.store.getChat(chatId);
}
if (!state.active) return;
const response = await this.runtimeFetch(`/v1/threads/${encodeURIComponent(state.threadId)}/events?since_seq=${state.active.seq}`, this.abort.signal);
if (!response.ok) throw new Error("Engine event stream unavailable");
for await (const event of readSse(response)) {
if (this.closed) return;
let record;
try { record = JSON.parse(event.data); } catch { continue; }
await this.consume(chatId, record);
if (!(await this.store.getChat(chatId)).active) return;
}
} catch {
if (this.closed) return;
const current = await this.store.getChat(chatId);
if (current.threadCreateSubmitted && !current.threadId) {
await this.enqueue(chatId, `${current.pendingAdmission.messageKey}:thread-uncertain`, "Engine thread creation is uncertain and will not be repeated. An operator must reconcile the original Engine request.");
return;
}
this.report("DingTalk Engine connection interrupted; retained receipts will be reconciled");
}
await delay(3000, undefined, { signal: this.abort.signal }).catch(() => {});
}
}
async consume(chatId, record) {
await this.mutate(async () => {
const state = await this.store.getChat(chatId);
if (!this.valid(state) || !state.active || !Number.isSafeInteger(record.seq) || record.seq <= state.active.seq) return;
const active = { ...state.active, seq: record.seq };
let outbox = state.outbox || [];
let terminal = false;
if (record.turn_id === active.turnId) {
const payload = record.payload || {};
const id = `${active.turnId}:${record.seq}`;
if (record.event === "item.delta" && payload.kind === "agent_message" && typeof payload.delta === "string") active.text += payload.delta;
if (active.text.length > 1000000) active.text = `${active.text.slice(0, 999900)}\n[Response exceeds the bridge text limit; inspect the Engine transcript.]`;
if (record.event === "approval.required") {
const approvalId = payload.id || payload.approval_id;
outbox = this.output({ outbox }, id, `Approval required: ${payload.tool_name || "tool"}\n${payload.description || ""}\n${approvalId ? `/allow ${approvalId}\n/deny ${approvalId}` : "No approval ID was provided; use the Engine client."}`);
}
if (record.event === "user_input.required") {
const inputId = payload.id || payload.input_id;
outbox = this.output({ outbox }, id, `${questionText(payload.request)}\n${inputId ? `/answer ${inputId} followed by your answers.` : "No question ID was provided; use the Engine client."}`);
}
terminal = record.event === "turn.completed" || (record.event === "turn.lifecycle" && ["failed", "canceled", "interrupted"].includes(payload.turn?.status || payload.status));
if (terminal) outbox = this.output({ outbox }, id, active.text || `Turn ${payload.turn?.status || payload.status || "completed"}.`);
}
await this.store.patchChat(chatId, { active: terminal ? null : active, outbox });
});
this.spawn(`output:${chatId}`, () => this.flush(chatId));
}
async flush(chatId) {
while (!this.closed) {
const next = await this.mutate(async () => {
const state = await this.store.getChat(chatId);
if (!this.valid(state)) return null;
const entry = (state.outbox || []).find((item) => item.status === "pending" || item.status === "awaiting_route");
if (!entry) return null;
try { replyRoute({ sessionWebhook: state.route?.url, sessionWebhookExpiredTime: state.route?.expiresAt }, this.now()); }
catch {
await this.store.patchChat(chatId, { outbox: state.outbox.map((item) => item.id === entry.id ? { ...item, status: "awaiting_route" } : item) });
return null;
}
await this.store.patchChat(chatId, { outbox: state.outbox.map((item) => item.id === entry.id ? { ...item, status: "sending" } : item) });
return { entry, route: state.route };
});
if (!next) return;
let confirmed = false;
try { await this.sendReply(next.route, next.entry.chunks[next.entry.index]); confirmed = true; } catch {}
await this.mutate(async () => {
const state = await this.store.getChat(chatId);
await this.store.patchChat(chatId, { outbox: state.outbox.map((item) => item.id !== next.entry.id ? item : {
...item, index: confirmed ? item.index + 1 : item.index,
status: confirmed ? item.index + 1 === item.chunks.length ? "sent" : "pending" : "uncertain"
}) });
});
}
}
async stop() {
this.closed = true;
this.abort.abort();
await Promise.allSettled([...this.tasks.values()]);
await this.serial;
}
}
+33 -1
View File
@@ -52,6 +52,7 @@ For first pairing, temporarily set `TELEGRAM_ALLOW_UNLISTED=true`, send the bot
- `/resume <thread_id>`
- `/model <name|default>`
- `/interrupt`
- `/answer <response>`
- `/compact`
- `/allow <approval_id> [remember]`
- `/deny <approval_id>`
@@ -75,6 +76,37 @@ Recovered deliveries require a saved, previously admitted sender identity that
still matches the current allowlist and group policy. Removing access and
restarting the bridge also stops recovery messages to that chat. Legacy state
without this identity stays detached; it does not cancel the runtime turn.
Use an authorized chat to inspect or interrupt that turn. A fresh admitted
Use the Runtime to control a turn without a recorded initiating human. A fresh admitted
message records identity for subsequent restart recovery. Keep allow-unlisted
mode off when operating with private workspace or runtime data.
## Durable admission and questions
The bridge saves each prompt, its Telegram message identity, initiating human,
and frozen Engine request in the existing thread map before advancing the
Telegram polling cursor. It uses Engine operation keys only when the Runtime
advertises both turn-operation idempotency and lookup. After an uncertain
submission, it looks up that original operation; a missing operation permits
retrying the same request and key, while an incomplete or unavailable receipt
stays retained. An Engine without these capabilities never receives an automatic
retry of an uncertain submission. `/status` reports retained admission state.
Changing the bot account, Runtime URL, allowlist, or group policy does not move
an existing pending request to another identity or Runtime.
A disconnected event stream reconnects to the accepted turn with its saved
cursor and accumulated reply. `/new`, `/resume`, and `/compact` cannot replace
a pending turn or retained final reply. Final replies are saved before sending.
An uncertain send is not automatically retried: the initiating human can retrieve
the saved reply with `/status`, which warns that the earlier delivery may have
succeeded. Telegram rate-limit refusals permit bounded retries. Progress updates
remain best effort; no send failure authorizes another Engine turn. This bridge does not provide a
cross-process lease for multiple instances sharing a state file. Run one bridge
process per bot token and state file.
When the Engine asks a question, reply with `/answer` followed by an option
number or text, or send the answer as a normal message. Answer multiple questions
on separate lines. The bridge checks the current Runtime question, turn, and
recorded initiating human before submitting an answer. Other group members
cannot answer, approve, or interrupt that person's turn. Group thread lists and
resume controls stay within that group's current conversation.
+285 -163
View File
@@ -1,5 +1,5 @@
import { createHash } from "node:crypto";
import {
activeTurnBlock,
activeTurnKeyboard,
approvalKeyboard,
callbackAction,
@@ -28,6 +28,9 @@ import {
telegramSendRetryDelayMs
} from "./lib.mjs";
import {
admitTurn,
questionText,
parseQuestionAnswers,
createRuntimeClient,
readJsonSafe,
readSse,
@@ -76,7 +79,9 @@ const config = {
turnTimeoutMs: Number(envFirst(process.env, "CODEWHALE_TURN_TIMEOUT_MS", "DEEPSEEK_TURN_TIMEOUT_MS") || 900000)
};
const { runtimeJson, authHeaders } = createRuntimeClient(config);
const { runtimeJson, authHeaders } = createRuntimeClient({ ...config, runtimeRequestTimeoutMs: 15000 });
const botAccountId = config.botToken.split(":")[0];
let runtimeCapabilities = {};
const threadStore = await ThreadStore.open(config.threadMapPath);
const activeTurnTasks = new Map();
@@ -101,6 +106,11 @@ if (!config.allowlist.length && !config.allowUnlisted) {
console.log("No allowlist configured. Incoming chats will receive their IDs and be refused.");
}
try {
runtimeCapabilities = (await runtimeJson("/v1/runtime/info"))?.capabilities || {};
} catch {
console.warn("Engine recovery capabilities unavailable; uncertain admissions will remain retained.");
}
await configureBotCommands().catch((error) => {
console.error("failed to configure Telegram bot command menu", error);
});
@@ -117,6 +127,7 @@ async function configureBotCommands() {
{ command: "threads", description: "List recent runtime threads" },
{ command: "new", description: "Create a new thread" },
{ command: "interrupt", description: "Interrupt the active turn" },
{ command: "answer", description: "Answer the current agent question" },
{ command: "compact", description: "Compact the current thread" },
{ command: "help", description: "Show command help" }
]
@@ -185,9 +196,6 @@ async function handleIncomingUpdate(update) {
if (!identity.chatId || !identity.messageId) return;
if (identity.isBot) return;
const messageKey = `${identity.chatId}:${identity.messageId}`;
if (await threadStore.recordMessage(messageKey)) return;
if (!identity.text) {
await sendText(identity.chatId, "Only text messages are supported in this first bridge.");
return;
@@ -214,8 +222,17 @@ async function handleIncomingUpdate(update) {
}
const command = parseCommand(scoped.text);
const messageKey = JSON.stringify([botAccountId, identity.chatId, identity.messageId]);
if (["prompt", "answer"].includes(commandAction(command).kind)) {
const state = await threadStore.getChat(identity.chatId);
if (state?.pendingAdmission?.messageKey === messageKey || state?.activeMessageKey === messageKey) {
await threadStore.save();
startTrackedTurnStream(identity.chatId);
return;
}
} else if (await threadStore.recordMessage(messageKey)) return;
await rememberAuthorizedIdentity(identity);
await handleCommand(identity.chatId, command, identity);
await handleCommand(identity.chatId, command, { ...identity, messageKey });
}
async function rememberAuthorizedIdentity({ chatId, chatType, userId, username, isBot }) {
@@ -231,6 +248,13 @@ async function isReplayCallbackUpdate(update) {
async function handleCommand(chatId, command, identity) {
const action = commandAction(command);
if (["new_thread", "resume", "compact"].includes(action.kind)) {
const state = await threadStore.getChat(chatId);
if (state?.pendingAdmission || state?.activeTurnId || state?.pendingReply) {
await sendText(chatId, "This chat has a pending turn or retained reply. Use /status to inspect it before replacing its thread.");
return;
}
}
switch (action.kind) {
case "help":
await sendText(chatId, helpText(), { replyMarkup: controlKeyboard() });
@@ -239,10 +263,10 @@ async function handleCommand(chatId, command, identity) {
await sendMenu(chatId);
return;
case "status":
await sendStatus(chatId);
await sendStatus(chatId, identity);
return;
case "threads":
await sendThreads(chatId);
await sendThreads(chatId, identity);
return;
case "new_thread": {
const state = await ensureThread(chatId, { forceNew: true });
@@ -250,10 +274,10 @@ async function handleCommand(chatId, command, identity) {
return;
}
case "resume":
await resumeThread(chatId, action.threadId);
await resumeThread(chatId, action.threadId, identity);
return;
case "interrupt":
await interruptActiveTurn(chatId);
await interruptActiveTurn(chatId, identity);
return;
case "compact":
await compactThread(chatId);
@@ -264,8 +288,12 @@ async function handleCommand(chatId, command, identity) {
case "set_model":
await setChatModel(chatId, action.modelName);
return;
case "answer":
await answerQuestion(chatId, action.text, identity);
await threadStore.recordMessage(identity.messageKey);
return;
case "prompt":
startPromptTurn(chatId, action.prompt, identity);
await startPromptTurn(chatId, action.prompt, identity);
return;
default:
await sendText(chatId, helpText(), { replyMarkup: controlKeyboard() });
@@ -311,15 +339,22 @@ async function handleCallbackQuery(query) {
}
async function handleModalAction(chatId, action, query = null, identity = null) {
if (["new_thread", "compact"].includes(action.kind)) {
const state = await threadStore.getChat(chatId);
if (state?.pendingAdmission || state?.activeTurnId || state?.pendingReply) {
await sendText(chatId, "This chat has a pending turn or retained reply. Use /status to inspect it before replacing its thread.");
return;
}
}
switch (action.kind) {
case "help":
await sendText(chatId, helpText(), { replyMarkup: controlKeyboard() });
return;
case "status":
await sendStatus(chatId);
await sendStatus(chatId, identity);
return;
case "threads":
await sendThreads(chatId);
await sendThreads(chatId, identity);
return;
case "new_thread": {
const state = await ensureThread(chatId, { forceNew: true });
@@ -327,7 +362,7 @@ async function handleModalAction(chatId, action, query = null, identity = null)
return;
}
case "interrupt":
await interruptActiveTurn(chatId);
await interruptActiveTurn(chatId, identity);
return;
case "compact":
await compactThread(chatId);
@@ -355,7 +390,7 @@ async function handleStoredAction(chatId, action, query = null, identity = null)
if (stored.kind === "resume") {
await threadStore.takeAction(action.token, owner);
await resumeThread(chatId, stored.threadId);
await resumeThread(chatId, stored.threadId, identity);
return;
}
@@ -396,14 +431,17 @@ async function sendMenu(chatId) {
);
}
async function ensureThread(chatId, { forceNew = false } = {}) {
async function ensureThread(chatId, { forceNew = false, threadRequest } = {}) {
const existing = await threadStore.getChat(chatId);
if (forceNew && (existing?.pendingAdmission || existing?.activeTurnId)) {
throw new Error("This chat has a pending turn or retained reply. Use /status to inspect it before replacing its thread.");
}
if (existing?.threadId && !forceNew) return existing;
const effectiveModel = existing?.model || config.model;
const thread = await runtimeJson("/v1/threads", {
method: "POST",
body: {
body: threadRequest || {
model: effectiveModel,
workspace: config.workspace,
mode: config.mode,
@@ -417,7 +455,7 @@ async function ensureThread(chatId, { forceNew = false } = {}) {
});
const state = {
...preservedChatStateFields(existing),
...preservedChatStateFields(existing, ["model", "authorizedIdentity", "pendingAdmission", "pendingReply"]),
threadId: thread.id,
lastSeq: 0,
activeTurnId: null,
@@ -427,63 +465,101 @@ async function ensureThread(chatId, { forceNew = false } = {}) {
return state;
}
function startPromptTurn(chatId, prompt, identity) {
if (activeTurnTasks.has(chatId)) {
void sendText(chatId, "Thread already has an active turn. Wait for it to finish or send /interrupt.", {
async function startPromptTurn(chatId, prompt, identity) {
const current = await threadStore.getChat(chatId);
if (current?.pendingQuestion) {
await answerQuestion(chatId, prompt, identity);
await threadStore.recordMessage(identity.messageKey);
return;
}
if (current?.pendingReply) {
await sendText(chatId, "The previous reply is retained because delivery was not confirmed. Its initiating human can retrieve it with /status before starting another turn.");
return;
}
if (activeTurnTasks.has(chatId) || current?.pendingAdmission || current?.activeTurnId) {
startTrackedTurnStream(chatId);
await sendText(chatId, "This chat has a pending turn. Use /status or /interrupt.", {
replyMarkup: activeTurnKeyboard()
}).catch((error) => {
console.error("failed to report active Telegram bridge turn", error);
});
return;
}
const controller = new AbortController();
const task = { controller };
activeTurnTasks.set(chatId, task);
void runPrompt(chatId, prompt, { signal: controller.signal, actorId: turnActor(identity) })
.catch((error) => {
console.error("failed to run Telegram bridge prompt", error);
})
.finally(() => {
if (activeTurnTasks.get(chatId) === task) {
activeTurnTasks.delete(chatId);
}
});
if (!prompt.trim()) return sendText(chatId, helpText(), { replyMarkup: controlKeyboard() });
const actorId = turnActor(identity);
if (!actorId || !identity.messageKey) throw new ApprovalOwnershipError("turn needs its initiating human and message");
const request = {
prompt, input_summary: prompt.slice(0, 200), model: current?.model || config.model,
mode: config.mode, allow_shell: config.allowShell, trust_mode: config.trustMode,
auto_approve: config.autoApprove
};
const operationKey = supportsAdmissionRecovery()
? createHash("sha256").update(identity.messageKey).digest("hex") : null;
if (operationKey) request.operation_key = operationKey;
const replay = await threadStore.recordMessage(identity.messageKey, chatId, { pendingAdmission: {
botAccountId, runtimeUrl: config.runtimeUrl, actorId, identity: {
chatId, chatType: identity.chatType, userId: identity.userId,
username: identity.username, isBot: identity.isBot
}, messageKey: identity.messageKey, request, operationKey,
threadId: current?.threadId || null, submitted: false, sinceSeq: null,
threadRequest: {
model: request.model, workspace: config.workspace, mode: request.mode,
allow_shell: request.allow_shell, trust_mode: request.trust_mode,
auto_approve: request.auto_approve, archived: false,
system_prompt: "You are being controlled from a Telegram phone chat. Keep status updates concise. Ask for tool approvals when needed; do not assume mobile messages imply blanket approval."
}
} });
if (!replay) startTrackedTurnStream(chatId);
}
function abortActiveTurnStreams() {
for (const task of activeTurnTasks.values()) {
task.controller?.abort();
}
for (const task of activeTurnTasks.values()) task.controller?.abort();
}
async function clearActiveTurn(chatId) {
await threadStore.patchChat(chatId, {
activeTurnId: null,
updatedAt: new Date().toISOString()
}).catch((error) => {
console.error("failed to clear Telegram bridge active turn", error);
activeTurnId: null, activeMessageKey: null, activeIdentity: null, activeBotAccountId: null, activeRuntimeUrl: null,
responseText: "", turnTerminal: false, pendingQuestion: null, updatedAt: new Date().toISOString()
});
}
function startTrackedTurnStream(chatId, threadId, turnId, sinceSeq) {
if (activeTurnTasks.has(chatId)) return false;
function recoveryAllowed(chatId, state) {
const pending = state?.pendingAdmission;
const identity = pending?.identity || state?.activeIdentity || state?.authorizedIdentity;
const accountId = pending?.botAccountId || state?.activeBotAccountId;
const runtimeUrl = pending?.runtimeUrl || state?.activeRuntimeUrl;
return (!runtimeUrl || runtimeUrl === config.runtimeUrl) && (!accountId || accountId === botAccountId) && identity && identity.chatId === chatId && !identity.isBot &&
["private", "group", "supergroup"].includes(identity.chatType) &&
(!isGroupChat(identity.chatType) || config.allowGroups) &&
isAllowed(identity, config.allowlist, config.allowUnlisted);
}
function supportsAdmissionRecovery() {
return runtimeCapabilities.turn_operation_idempotency === true && runtimeCapabilities.turn_operation_lookup === true;
}
function startTrackedTurnStream(chatId) {
if (stopping || activeTurnTasks.has(chatId)) return false;
const controller = new AbortController();
const task = { controller };
activeTurnTasks.set(chatId, task);
void streamTurnEvents(chatId, threadId, turnId, sinceSeq, { signal: controller.signal })
.catch((error) => {
console.error("failed to stream Telegram bridge turn", error);
})
.finally(async () => {
if (activeTurnTasks.get(chatId) === task) {
activeTurnTasks.delete(chatId);
void (async () => {
while (!stopping && !controller.signal.aborted) {
const state = await threadStore.getChat(chatId);
if (!recoveryAllowed(chatId, state)) return;
if (!state?.pendingAdmission && !state?.activeTurnId) return;
try {
if (await runPrompt(chatId, { signal: controller.signal })) return;
const pending = (await threadStore.getChat(chatId))?.pendingAdmission;
if (pending?.submitted && (!pending.operationKey || !supportsAdmissionRecovery())) return;
} catch (error) {
console.error("Telegram turn recovery deferred; original request retained", error);
}
if (!stopping) {
await clearActiveTurn(chatId);
}
});
if (!stopping && !controller.signal.aborted) await delay(2000);
}
})().catch((error) => {
console.error("Telegram recovery stopped; persisted state retained", error);
}).finally(() => {
if (activeTurnTasks.get(chatId) === task) activeTurnTasks.delete(chatId);
});
return true;
}
@@ -491,103 +567,48 @@ function turnActor(identity) {
return identity?.userId && !identity.isBot ? `telegram-user:${identity.userId}` : "";
}
async function runPrompt(chatId, prompt, options = {}) {
if (!prompt.trim()) {
await sendText(chatId, helpText(), { replyMarkup: controlKeyboard() });
return;
}
if (!options.actorId) throw new ApprovalOwnershipError("turn needs its initiating human");
const state = await ensureThread(chatId);
const effectiveModel = state?.model || config.model;
const detail = await runtimeJson(`/v1/threads/${encodeURIComponent(state.threadId)}`);
const activeBlock = activeTurnBlock(detail, state);
if (activeBlock) {
await threadStore.patchChat(chatId, {
activeTurnId: activeBlock.turnId,
async function runPrompt(chatId, options = {}) {
let state = await threadStore.getChat(chatId);
if (!recoveryAllowed(chatId, state)) return true;
let pending = state?.pendingAdmission;
if (pending) {
const admission = await admitTurn({
runtimeJson, pending, capabilities: runtimeCapabilities,
persistPending: (value) => threadStore.patchChat(chatId, { pendingAdmission: value }),
ensureThread: async (threadRequest) => (await ensureThread(chatId, { threadRequest })).threadId
});
pending = admission.pending;
const turn = admission.turn;
if (!turn) {
if (admission.blocked && !pending.submitted) {
await threadStore.patchChat(chatId, { pendingAdmission: null });
await sendText(chatId, "Thread already has an active turn. Wait for it to finish or send /interrupt.", { replyMarkup: activeTurnKeyboard() });
return true;
}
return admission.uncertain === true;
}
await threadStore.recordTurnOrigin(chatId, pending.threadId, turn.id, pending.actorId);
state = await threadStore.patchChat(chatId, {
pendingAdmission: null, activeTurnId: turn.id, activeMessageKey: pending.messageKey,
activeIdentity: pending.identity, activeBotAccountId: pending.botAccountId,
activeRuntimeUrl: pending.runtimeUrl,
lastSeq: pending.sinceSeq, responseText: "", turnTerminal: false,
updatedAt: new Date().toISOString()
});
await sendText(chatId, activeBlock.message, { replyMarkup: activeTurnKeyboard() });
return;
await sendTurnText(chatId, `Started turn ${turn.id}`, { replyMarkup: activeTurnKeyboard() });
}
if (state.activeTurnId) {
await threadStore.patchChat(chatId, { activeTurnId: null });
}
const sinceSeq = Number(detail.latest_seq || state.lastSeq || 0);
const turnResponse = await runtimeJson(
`/v1/threads/${encodeURIComponent(state.threadId)}/turns`,
{
method: "POST",
body: {
prompt,
input_summary: prompt.slice(0, 200),
model: effectiveModel,
mode: config.mode,
allow_shell: config.allowShell,
trust_mode: config.trustMode,
auto_approve: config.autoApprove
}
}
);
const turnId = turnResponse.turn?.id;
// A turn the Runtime accepted keeps streaming even when its origin cannot be
// recorded; its approvals then fail closed and are decided from the TUI.
if (turnId && options.actorId) {
await threadStore.recordTurnOrigin(chatId, state.threadId, turnId, options.actorId);
}
await threadStore.patchChat(chatId, {
activeTurnId: turnId || null,
lastSeq: sinceSeq,
updatedAt: new Date().toISOString()
});
await sendTurnText(chatId, `Started turn ${turnId || "(unknown)"}`, {
replyMarkup: activeTurnKeyboard()
});
try {
await streamTurnEvents(chatId, state.threadId, turnId, sinceSeq, options);
} finally {
if (!stopping) {
await clearActiveTurn(chatId);
}
if (!state?.activeTurnId) return true;
if (state.turnTerminal || await streamTurnEvents(chatId, state.threadId, state.activeTurnId, Number(state.lastSeq || 0), options)) {
await clearActiveTurn(chatId);
return true;
}
return false;
}
async function reattachActiveTurns() {
for (const [chatId, state] of threadStore.listChats()) {
if (!state?.threadId || !state.activeTurnId) continue;
const identity = state.authorizedIdentity;
// Legacy state has no verified sender provenance. Never recover delivery
// under a previous allowlist or group policy, even to report completion.
if (!identity || identity.chatId !== chatId || identity.isBot ||
!["private", "group", "supergroup"].includes(identity.chatType) ||
(isGroupChat(identity.chatType) && !config.allowGroups) ||
!isAllowed(identity, config.allowlist, config.allowUnlisted)) continue;
const detail = await runtimeJson(`/v1/threads/${encodeURIComponent(state.threadId)}`);
const runningTurn = latestRunningTurn(detail);
if (!runningTurn) {
await threadStore.patchChat(chatId, {
activeTurnId: null,
lastSeq: Number(detail.latest_seq || state.lastSeq || 0),
updatedAt: new Date().toISOString()
});
await sendText(chatId, `Bridge restarted. No active turn remains for ${state.threadId}.`);
continue;
}
const turnId = runningTurn.id || state.activeTurnId;
const sinceSeq = Number(state.lastSeq || 0);
await threadStore.patchChat(chatId, {
activeTurnId: turnId,
updatedAt: new Date().toISOString()
});
await sendTurnText(
chatId,
`Bridge restarted. Reattaching to active turn ${turnId} from seq ${sinceSeq}.`
);
startTrackedTurnStream(chatId, state.threadId, turnId, sinceSeq);
if ((!state?.pendingAdmission && !state?.activeTurnId) || !recoveryAllowed(chatId, state)) continue;
startTrackedTurnStream(chatId);
}
}
@@ -604,7 +625,8 @@ async function streamTurnEvents(chatId, threadId, turnId, sinceSeq, options = {}
} else {
options.signal?.addEventListener("abort", abortFromCaller, { once: true });
}
let responseText = "";
let responseText = (await threadStore.getChat(chatId))?.responseText || "";
let turnTerminal = false;
let latestSeq = sinceSeq;
let flushedSeq = sinceSeq;
let lastSeqFlushAt = 0;
@@ -615,7 +637,7 @@ async function streamTurnEvents(chatId, threadId, turnId, sinceSeq, options = {}
async function flushLastSeq(force = false) {
if (latestSeq <= flushedSeq) return;
if (!force && Date.now() - lastSeqFlushAt < LAST_SEQ_FLUSH_INTERVAL_MS) return;
await threadStore.patchChat(chatId, { lastSeq: latestSeq });
await threadStore.patchChat(chatId, { lastSeq: latestSeq, responseText, turnTerminal });
flushedSeq = latestSeq;
lastSeqFlushAt = Date.now();
}
@@ -654,7 +676,6 @@ async function streamTurnEvents(chatId, threadId, turnId, sinceSeq, options = {}
if (!event.data) continue;
const record = JSON.parse(event.data);
latestSeq = Math.max(latestSeq, Number(record.seq || 0));
await flushLastSeq(false);
if (turnId && record.turn_id && record.turn_id !== turnId) continue;
const lifecycleStatus =
@@ -664,7 +685,7 @@ async function streamTurnEvents(chatId, threadId, turnId, sinceSeq, options = {}
const stopTypingEvent =
record.event === "turn.completed" ||
["failed", "canceled", "interrupted"].includes(lifecycleStatus);
if (typingPaused && record.event !== "approval.required" && !stopTypingEvent) {
if (typingPaused && !["approval.required", "user_input.required"].includes(record.event) && !stopTypingEvent) {
typingPaused = false;
void tickTyping();
}
@@ -679,6 +700,22 @@ async function streamTurnEvents(chatId, threadId, turnId, sinceSeq, options = {}
}
}
if (record.event === "user_input.required") {
typingPaused = true;
const input = record.payload || {};
const inputId = input.input_id || input.id;
if (inputId && input.request) {
await threadStore.patchChat(chatId, { pendingQuestion: { inputId, request: input.request, turnId } });
await sendTurnText(chatId, `${questionText(input.request)}\nUse /answer followed by your response.`, { replyMarkup: activeTurnKeyboard() });
}
}
if (["user_input.answered", "user_input.canceled"].includes(record.event)) {
const state = await threadStore.getChat(chatId);
if (state?.pendingQuestion?.inputId === (record.payload?.input_id || record.payload?.id)) {
await threadStore.patchChat(chatId, { pendingQuestion: null });
}
}
if (record.event === "approval.required") {
typingPaused = true;
const approval = record.payload || {};
@@ -732,28 +769,31 @@ async function streamTurnEvents(chatId, threadId, turnId, sinceSeq, options = {}
const error = turn.error ? `\n${turn.error}` : "";
if (status !== "completed") {
await sendTurnText(chatId, `Turn ${status}.${error}`.trim(), {
replyMarkup: controlKeyboard()
replyMarkup: controlKeyboard(), retain: true
});
} else {
await sendTurnText(chatId, responseText.trim() || "Turn completed.", {
replyMarkup: controlKeyboard()
replyMarkup: controlKeyboard(), retain: true
});
}
return;
turnTerminal = true;
return true;
}
if (record.event === "turn.lifecycle") {
if (["failed", "canceled", "interrupted"].includes(lifecycleStatus)) {
typingPaused = true;
await sendTurnText(chatId, `Turn ${lifecycleStatus}.`, { replyMarkup: controlKeyboard() });
return;
await sendTurnText(chatId, `Turn ${lifecycleStatus}.`, { replyMarkup: controlKeyboard(), retain: true });
turnTerminal = true;
return true;
}
}
await flushLastSeq(false);
}
} catch (error) {
if (error.name === "AbortError") {
if (timedOut) {
await sendTurnText(chatId, `Turn timed out after ${Math.round(config.turnTimeoutMs / 1000)}s.`);
await sendTurnText(chatId, "Observation timed out; the bridge will reconnect to the same turn.");
} else if (!stopping) {
await sendTurnText(chatId, "Turn stream aborted.");
}
@@ -768,7 +808,47 @@ async function streamTurnEvents(chatId, threadId, turnId, sinceSeq, options = {}
}
}
async function sendStatus(chatId) {
async function answerQuestion(chatId, text, identity) {
const state = await threadStore.getChat(chatId);
if (!state?.threadId || !state.activeTurnId || !recoveryAllowed(chatId, state) ||
threadStore.turnOrigin(chatId, state.threadId, state.activeTurnId)?.actorId !== turnActor(identity)) {
await sendText(chatId, "Only the recorded initiating human can answer this turn's question.");
return;
}
const detail = await runtimeJson(`/v1/threads/${encodeURIComponent(state.threadId)}`);
const inputs = (detail.pending_user_inputs || []).filter((input) => input.turn_id === state.activeTurnId);
const input = state.pendingQuestion
? inputs.find((candidate) => (candidate.id || candidate.input_id) === state.pendingQuestion.inputId)
: inputs.length === 1 ? inputs[0] : null;
const inputId = input?.id || input?.input_id;
if (!inputId) {
await threadStore.patchChat(chatId, { pendingQuestion: null });
await sendText(chatId, "No matching question is waiting in this turn.");
return;
}
const answers = parseQuestionAnswers(input.request, text);
if (!answers) {
await sendText(chatId, questionText(input.request));
return;
}
await runtimeJson(`/v1/user-input/${encodeURIComponent(state.threadId)}/${encodeURIComponent(inputId)}`, {
method: "POST", body: { answers }
});
if ((await threadStore.getChat(chatId))?.pendingQuestion?.inputId === inputId) {
await threadStore.patchChat(chatId, { pendingQuestion: null });
}
await sendText(chatId, "Answer sent.");
}
async function sendStatus(chatId, identity) {
const state = await threadStore.getChat(chatId);
const reply = state?.pendingReply;
if (reply && recoveryAllowed(chatId, { pendingAdmission: reply }) && reply.actorId === turnActor(identity)) {
await sendText(chatId, `Retained reply (the earlier delivery may have succeeded):\n\n${reply.text}`, { replyMarkup: controlKeyboard() });
if ((await threadStore.getChat(chatId))?.pendingReply?.key === reply.key) {
await threadStore.patchChat(chatId, { pendingReply: null });
}
}
const [health, runtimeInfo, workspace] = await Promise.all([
runtimeJson("/health", { auth: false }),
runtimeJson("/v1/runtime/info"),
@@ -784,7 +864,11 @@ async function sendStatus(chatId) {
`workspace=${workspace.workspace}`,
`git_repo=${workspace.git_repo}`,
workspace.branch ? `branch=${workspace.branch}` : "",
`staged=${workspace.staged} unstaged=${workspace.unstaged} untracked=${workspace.untracked}`
`staged=${workspace.staged} unstaged=${workspace.unstaged} untracked=${workspace.untracked}`,
state?.pendingAdmission ? "The original prompt is retained while Engine admission is reconciled." : "",
state?.pendingAdmission?.submitted && !state.pendingAdmission.operationKey ? "This Engine cannot reconcile the uncertain admission; inspect the Runtime before repeating the prompt." : "",
state?.activeTurnId ? `turn=${state.activeTurnId}` : "",
state?.pendingQuestion ? questionText(state.pendingQuestion.request) : ""
]
.filter(Boolean)
.join("\n"),
@@ -792,8 +876,11 @@ async function sendStatus(chatId) {
);
}
async function sendThreads(chatId) {
const threads = await runtimeJson("/v1/threads/summary?limit=8&include_archived=true");
async function sendThreads(chatId, identity) {
const state = await threadStore.getChat(chatId);
const threads = isGroupChat(identity?.chatType)
? state?.threadId ? [{ id: state.threadId }] : []
: await runtimeJson("/v1/threads/summary?limit=8&include_archived=true");
if (!threads.length) {
await sendText(chatId, "No runtime threads yet.", { replyMarkup: controlKeyboard() });
return;
@@ -818,14 +905,23 @@ async function sendThreads(chatId) {
);
}
async function resumeThread(chatId, args) {
async function resumeThread(chatId, args, identity) {
const existing = await threadStore.getChat(chatId);
if (existing?.pendingAdmission || existing?.activeTurnId || existing?.pendingReply) {
await sendText(chatId, "This chat has a pending turn or retained reply. Use /status to inspect it before replacing its thread.");
return;
}
const threadId = args.trim();
if (!threadId) {
await sendText(chatId, "Usage: /resume <thread_id>");
return;
}
if (isGroupChat(identity?.chatType) && (threadId !== existing?.threadId ||
!(existing?.turnOrigins || []).some((origin) => origin.threadId === threadId && origin.actorId === turnActor(identity)))) {
await sendText(chatId, "Group chats can only resume their own thread through a recorded initiating human.");
return;
}
const detail = await runtimeJson(`/v1/threads/${encodeURIComponent(threadId)}`);
const existing = await threadStore.getChat(chatId);
await threadStore.setChat(chatId, {
...preservedChatStateFields(existing),
threadId,
@@ -836,8 +932,17 @@ async function resumeThread(chatId, args) {
await sendText(chatId, `Resumed thread ${threadId}`, { replyMarkup: controlKeyboard() });
}
async function interruptActiveTurn(chatId) {
async function interruptActiveTurn(chatId, identity) {
const state = await threadStore.getChat(chatId);
if (state?.pendingAdmission) {
if (state.pendingAdmission.actorId !== turnActor(identity)) {
await sendText(chatId, "Only the initiating human can control this pending turn.");
return;
}
startTrackedTurnStream(chatId);
await sendText(chatId, "The original prompt's admission is still being reconciled. Use /status, then /interrupt once its turn is known.");
return;
}
if (!state?.threadId) {
await sendText(chatId, "No runtime thread recorded for this chat.");
return;
@@ -849,6 +954,10 @@ async function interruptActiveTurn(chatId) {
await sendText(chatId, "No active turn recorded for this chat.");
return;
}
if (threadStore.turnOrigin(chatId, state.threadId, turnId)?.actorId !== turnActor(identity)) {
await sendText(chatId, "Only the recorded initiating human can interrupt this turn here. Use the Runtime for an unbound turn.");
return;
}
await runtimeJson(
`/v1/threads/${encodeURIComponent(state.threadId)}/turns/${encodeURIComponent(
turnId
@@ -959,8 +1068,21 @@ async function sendTypingAction(chatId) {
}
async function sendTurnText(chatId, text, options = {}) {
let reply;
if (options.retain) {
const state = await threadStore.getChat(chatId);
reply = {
key: createHash("sha256").update(JSON.stringify([state.threadId, state.activeTurnId, text])).digest("hex"),
text, identity: state.activeIdentity, actorId: turnActor(state.activeIdentity),
botAccountId: state.activeBotAccountId, runtimeUrl: state.activeRuntimeUrl
};
await threadStore.patchChat(chatId, { pendingReply: reply, turnTerminal: true });
}
try {
await sendText(chatId, text, options);
if (reply && (await threadStore.getChat(chatId))?.pendingReply?.key === reply.key) {
await threadStore.patchChat(chatId, { pendingReply: null });
}
} catch (error) {
console.error("failed to send Telegram turn update", error);
}
@@ -1002,7 +1124,7 @@ async function telegramApiOnce(method, body = {}, options = {}) {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify(body),
signal: options.signal
signal: options.signal || AbortSignal.timeout(method === "getUpdates" ? (config.pollTimeoutSeconds + 15) * 1000 : 15000)
});
const payload = await readJsonSafe(response);
if (!response.ok || payload?.ok === false) {
+4 -22
View File
@@ -84,10 +84,13 @@ export function stripGroupPrefix(text, { chatType, requirePrefix, prefix }) {
}
export function parseCommand(text) {
const answer = String(text || "").trim().match(/^\/answer(?:@\w+)?(?:\s+([\s\S]*))?$/i);
if (answer) return { name: "answer", args: (answer[1] || "").trim() };
return coreParseCommand(text, { stripBotMention: true });
}
export function commandAction(command) {
if (command.name === "answer") return { kind: "answer", text: command.args };
return coreCommandAction(command, { allowMenu: true, allowStart: true });
}
@@ -404,31 +407,9 @@ export function telegramSendRetryDelayMs(error, attempt = 0) {
}
return 3000;
}
if (isTransientTelegramSendError(error) && attempt < 2) {
return attempt === 0 ? 1000 : 2000;
}
return null;
}
function isTransientTelegramSendError(error) {
if (!error || error.errorCode) return false;
const name = String(error.name || "");
if (name === "AbortError" || name === "TimeoutError") return false;
if (error instanceof TypeError) return true;
const code = String(error.code || error.cause?.code || "");
if (["ECONNRESET", "ECONNREFUSED", "EAI_AGAIN", "ENOTFOUND", "ETIMEDOUT"].includes(code)) {
return true;
}
const message = String(error.message || "").toLowerCase();
return (
message.includes("fetch failed") ||
message.includes("network") ||
message.includes("socket hang up")
);
}
export function looksLikePollingConflict(error) {
const text = String(error?.description || error?.message || "").toLowerCase();
return error?.errorCode === 409 || text.includes("terminated by other getupdates request");
@@ -578,6 +559,7 @@ export function helpText() {
"/compact - compact the current thread",
"/allow <approval_id> [remember] - approve a pending tool call",
"/deny <approval_id> - deny a pending tool call",
"/answer <response> - answer the current agent question",
"",
"Anything else is sent as a CodeWhale prompt."
].join("\n");
@@ -34,19 +34,19 @@ function extractFunction(source, name) {
assert.fail(`${name} body should close`);
}
test("prompt command starts a tracked background turn instead of blocking update dispatch", async () => {
test("prompt command persists admission before starting background recovery", async () => {
const source = await readBridgeSource();
const handleCommand = extractFunction(source, "handleCommand");
const promptCase = handleCommand.slice(handleCommand.indexOf('case "prompt":'));
assert.match(source, /const activeTurnTasks = new Map\(\);/);
assert.match(promptCase, /startPromptTurn\(chatId, action\.prompt, identity\);/);
assert.match(promptCase, /await startPromptTurn\(chatId, action\.prompt, identity\);/);
assert.doesNotMatch(promptCase, /await\s+runPrompt\(/);
const starter = extractFunction(source, "startPromptTurn");
assert.ok(
starter.indexOf("activeTurnTasks.set(chatId") < starter.indexOf("void runPrompt"),
"turn registry entry must be installed before runPrompt can await"
starter.indexOf("await threadStore.recordMessage") < starter.lastIndexOf("startTrackedTurnStream(chatId)"),
"prompt and inbound receipt must be durable before admission starts"
);
});
@@ -107,7 +107,7 @@ test("callback replay is ignored before modal dispatch", async () => {
assert.match(replayHelper, /return threadStore\.recordMessage\(`callback:\$\{update\.update_id\}`\);/);
assert.ok(
resumeCase.indexOf("await threadStore.takeAction(action.token, owner);") <
resumeCase.indexOf("await resumeThread(chatId, stored.threadId);"),
resumeCase.indexOf("await resumeThread(chatId, stored.threadId, identity);"),
"resume callback actions should be consumed before dispatch"
);
});
@@ -117,13 +117,15 @@ test("reattached streams are detached and shutdown preserves active turn state",
const reattach = extractFunction(source, "reattachActiveTurns");
const runPrompt = extractFunction(source, "runPrompt");
assert.match(reattach, /startTrackedTurnStream\(chatId, state\.threadId, turnId, sinceSeq\);/);
assert.match(reattach, /startTrackedTurnStream\(chatId\);/);
assert.doesNotMatch(reattach, /await\s+streamTurnEvents\(/);
assert.match(source, /async function clearActiveTurn\(chatId\)/);
assert.match(runPrompt, /if \(!stopping\) {\s*await clearActiveTurn\(chatId\);\s*}/);
assert.match(runPrompt, /if \(state\.turnTerminal \|\| await streamTurnEvents/);
assert.match(runPrompt, /await clearActiveTurn\(chatId\);/);
const trackedStream = extractFunction(source, "startTrackedTurnStream");
assert.match(trackedStream, /if \(!stopping\) {\s*await clearActiveTurn\(chatId\);\s*}/);
assert.doesNotMatch(trackedStream, /clearActiveTurn/);
assert.match(trackedStream, /while \(!stopping && !controller\.signal\.aborted\)/);
});
test("turn update sends retry without ending the stream", async () => {
@@ -151,7 +153,7 @@ test("turn streams keep Telegram typing visible and pause while waiting for appr
assert.match(streamTurnEvents, /const typingTimer = setInterval\(\(\) => {\s*void tickTyping\(\);/);
assert.match(streamTurnEvents, /void tickTyping\(\);/);
assert.match(streamTurnEvents, /const stopTypingEvent =/);
assert.match(streamTurnEvents, /if \(typingPaused && record\.event !== "approval\.required" && !stopTypingEvent\)/);
assert.match(streamTurnEvents, /if \(typingPaused && !\["approval\.required", "user_input\.required"\]\.includes\(record\.event\) && !stopTypingEvent\)/);
assert.match(streamTurnEvents, /typingPaused = true;/);
assert.match(streamTurnEvents, /clearInterval\(typingTimer\);/);
assert.match(sendTypingAction, /telegramApi\(\s*"sendChatAction"/);
@@ -169,13 +171,13 @@ test("turn streams debounce last-seq writes and flush before exit", async () =>
assert.match(source, /const LAST_SEQ_FLUSH_INTERVAL_MS = 2000;/);
assert.doesNotMatch(
streamWithoutFlushHelper,
/await threadStore\.patchChat\(chatId, \{ lastSeq: latestSeq \}\);/
/await threadStore\.patchChat\(chatId, \{ lastSeq: latestSeq, responseText, turnTerminal \}\);/
);
assert.match(streamTurnEvents, /await flushLastSeq\(false\);/);
assert.match(streamTurnEvents, /await flushLastSeq\(true\);/);
assert.match(flushLastSeq, /if \(latestSeq <= flushedSeq\) return;/);
assert.match(flushLastSeq, /Date\.now\(\) - lastSeqFlushAt < LAST_SEQ_FLUSH_INTERVAL_MS/);
assert.match(flushLastSeq, /await threadStore\.patchChat\(chatId, \{ lastSeq: latestSeq \}\);/);
assert.match(flushLastSeq, /await threadStore\.patchChat\(chatId, \{ lastSeq: latestSeq, responseText, turnTerminal \}\);/);
assert.match(flushLastSeq, /flushedSeq = latestSeq;/);
});
@@ -309,8 +309,8 @@ test("telegramSendRetryDelayMs retries only safe send failures", () => {
telegramSendRetryDelayMs({ errorCode: 429, parameters: { retry_after: 3 } }, 3),
null
);
assert.equal(telegramSendRetryDelayMs(new TypeError("fetch failed"), 0), 1000);
assert.equal(telegramSendRetryDelayMs(new TypeError("fetch failed"), 1), 2000);
assert.equal(telegramSendRetryDelayMs(new TypeError("fetch failed"), 0), null);
assert.equal(telegramSendRetryDelayMs(new TypeError("fetch failed"), 1), null);
assert.equal(telegramSendRetryDelayMs(new TypeError("fetch failed"), 2), null);
assert.equal(telegramSendRetryDelayMs({ name: "AbortError" }, 0), null);
assert.equal(telegramSendRetryDelayMs({ errorCode: 500 }, 0), null);
+1 -41
View File
@@ -170,47 +170,7 @@ export function approvalText(approval, code) {
].filter(Boolean).join("\n");
}
export function questionText(request) {
const questions = Array.isArray(request?.questions) ? request.questions : [];
const lines = ["需要你回答 / Codewhale is asking"];
for (const [index, question] of questions.entries()) {
const prefix = questions.length > 1 ? `${index + 1}. ` : "";
lines.push(`${prefix}${question.header ? `${question.header}: ` : ""}${question.question || ""}`);
for (const [optionIndex, option] of (question.options || []).entries()) {
lines.push(` ${optionIndex + 1}) ${option.label}${option.description ? ` - ${clip(option.description, 120)}` : ""}`);
}
}
lines.push(questions.length > 1
? "每行回答一个问题,可回复选项编号或文字。/ Answer one question per line, by option number or in words."
: "回复选项编号或文字。/ Reply with an option number or in words.");
return lines.join("\n");
}
export function parseQuestionAnswers(request, text) {
const questions = Array.isArray(request?.questions) ? request.questions : [];
if (!questions.length) return null;
const replies = questions.length > 1
? String(text || "").split(/\n+/).map((line) => line.trim()).filter(Boolean)
: [String(text || "").trim()];
if (replies.length !== questions.length || replies.some((reply) => !reply)) return null;
const answers = [];
for (const [index, question] of questions.entries()) {
const reply = replies[index];
const options = question.options || [];
const picks = question.multi_select ? reply.split(/[\s,,、]+/u).filter(Boolean) : [reply];
const chosen = picks.every((pick) => /^\d+$/.test(pick) && Number(pick) >= 1 && Number(pick) <= options.length)
? picks.map((pick) => options[Number(pick) - 1])
: options.filter((option) => option.label.toLowerCase() === reply.toLowerCase());
if (chosen.length) {
for (const option of chosen) answers.push({ id: question.id, label: option.label, value: option.label });
} else if (question.allow_free_text || !options.length) {
answers.push({ id: question.id, label: "Other", value: reply });
} else {
return null;
}
}
return answers;
}
export { questionText, parseQuestionAnswers } from "../../bridge-core/src/lib.mjs";
export function isSessionExpired(response) {
return response?.ret === -14 || response?.errcode === -14;
+133
View File
@@ -0,0 +1,133 @@
import { createHash } from 'node:crypto';
const SHA = /^[a-f0-9]{40}(?:[a-f0-9]{24})?$/i;
const REPO = /^[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+$/;
const MAX_BYTES = 2 * 1024 * 1024;
function requireValue(valid) {
if (!valid) throw new Error('invalid_findings_artifact');
}
function text(value, max = 20000) {
requireValue(typeof value === 'string' && value.length <= max && !/[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/.test(value));
return value;
}
export function repositoryPath(value) {
return typeof value === 'string' && value.length > 0 && value.length <= 4096
&& !/[\\\u0000-\u001f\u007f]/.test(value) && !value.startsWith('/')
&& !value.split('/').some(part => !part || part === '.' || part === '..');
}
function location(value) {
requireValue(value.path == null || repositoryPath(value.path));
requireValue(value.line == null || (Number.isSafeInteger(value.line) && value.line > 0));
return { path: value.path ?? null, line: value.line ?? null };
}
function list(value, convert, max = 200) {
requireValue(Array.isArray(value) && value.length <= max);
return value.map(convert);
}
function review(value) {
requireValue(value && typeof value === 'object');
const result = {
summary: text(value.summary),
issues: list(value.issues, issue => {
requireValue(issue && ['blocker', 'concern', 'nit', 'error', 'warning', 'info', 'high', 'medium', 'low', 'critical'].includes(issue.severity));
requireValue(issue.authorship == null || ['introduced', 'preexisting'].includes(issue.authorship));
requireValue(issue.inline_anchorable == null || typeof issue.inline_anchorable === 'boolean');
const item = { ...location(issue), severity: issue.severity, title: text(issue.title, 1000),
description: text(issue.description), authorship: issue.authorship ?? null,
inline_anchorable: issue.inline_anchorable ?? null };
if (issue.evidence != null) {
requireValue(['reproduced', 'traced', 'suspected'].includes(issue.evidence.tier));
item.evidence = { tier: issue.evidence.tier, detail: text(issue.evidence.detail) };
}
if (issue.contract != null) item.contract = { became: text(issue.contract.became),
dependents_checked: list(issue.contract.dependents_checked, item => text(item, 4096), 100) };
if (issue.kind != null) {
requireValue(['defect', 'intent', 'contract'].includes(issue.kind));
item.kind = issue.kind;
}
return item;
}),
suggestions: list(value.suggestions, item => ({ ...location(item), suggestion: text(item.suggestion) })),
overall_assessment: text(value.overall_assessment),
verified: list(value.verified ?? [], item => text(item), 100),
not_reviewed: list(value.not_reviewed ?? [], item => text(item), 100),
ruled_out: list(value.ruled_out ?? [], item => text(item), 100),
};
if (value.merge_risk != null) {
requireValue(['low', 'medium', 'high'].includes(value.merge_risk.level));
result.merge_risk = { level: value.merge_risk.level, reason: text(value.merge_risk.reason) };
}
return result;
}
export function validateArtifact(value) {
requireValue(value?.schema_version === 1 && value.kind === 'codewhale-review-findings');
requireValue(REPO.test(value.repository || '') && Number.isSafeInteger(value.pr_number) && value.pr_number > 0);
requireValue(SHA.test(value.base_sha || '') && SHA.test(value.head_sha || ''));
requireValue(/^[1-9][0-9]*$/.test(value.run_id || '') && /^[1-9][0-9]*$/.test(value.run_attempt || ''));
requireValue(value.complete === true && value.publication === 'not_requested');
requireValue(Number.isSafeInteger(value.review_passes) && value.review_passes > 0 && value.review_passes <= 64);
const result = { schema_version: 1, kind: value.kind, repository: value.repository, pr_number: value.pr_number,
base_sha: value.base_sha, head_sha: value.head_sha, run_id: value.run_id, run_attempt: value.run_attempt,
complete: true, publication: 'not_requested', review_passes: value.review_passes, review: review(value.review) };
requireValue(Buffer.byteLength(JSON.stringify(result)) <= MAX_BYTES);
return result;
}
export function parseArtifact(raw) {
requireValue(Buffer.byteLength(raw) <= MAX_BYTES);
return validateArtifact(JSON.parse(raw));
}
export function findingsArtifact(data, receipt, env) {
const secrets = ['GH_TOKEN', 'CODEWHALE_API_KEY', 'DEEPSEEK_API_KEY', 'ANTHROPIC_API_KEY',
'OPENROUTER_API_KEY', 'ZAI_API_KEY', 'MODELSTUDIO_API_KEY'].map(key => env[key]).filter(Boolean);
const clean = value => typeof value === 'string'
? secrets.reduce((result, secret) => result.split(secret).join('[REDACTED]'), value)
: Array.isArray(value) ? value.map(clean)
: value && typeof value === 'object' ? Object.fromEntries(Object.entries(value).map(([key, item]) => [key, clean(item)])) : value;
return validateArtifact({ schema_version: 1, kind: 'codewhale-review-findings', repository: receipt.repository,
pr_number: receipt.pr_number, base_sha: receipt.base_sha, head_sha: receipt.head_sha,
run_id: env.GITHUB_RUN_ID, run_attempt: env.GITHUB_RUN_ATTEMPT, complete: true,
publication: 'not_requested', review_passes: data.review_passes, review: clean(review(data.review)) });
}
export function fingerprint(value) {
return createHash('sha256').update(JSON.stringify(value)).digest('hex');
}
export function markdown(value) {
return value.replaceAll('&', '&amp;').replaceAll('<', '&lt;').replaceAll('>', '&gt;')
.replaceAll('@', '&#64;').replace(/[\\`*_{}\[\]()!|#~]/g, '\\$&');
}
export function issueBody(issue) {
let body = `**[${issue.severity.toUpperCase()}] ${markdown(issue.title)}**\n\n${markdown(issue.description)}`;
if (issue.evidence) body += `\n\n${issue.evidence.tier}: ${markdown(issue.evidence.detail)}`;
if (issue.contract) body += `\n\nContract: ${markdown(issue.contract.became)}\n\nDependents checked: ${issue.contract.dependents_checked.map(markdown).join(', ')}`;
return body;
}
export function renderSummary(value, issues, duplicateCount) {
const result = ['## Codewhale review', `Reviewed commit: ${value.head_sha}`, markdown(value.review.summary)];
if (value.review.merge_risk) result.push(`**Merge risk: ${value.review.merge_risk.level}** — ${markdown(value.review.merge_risk.reason)}`);
for (const issue of issues) {
const location = issue.path ? `\n\nLocation: ${markdown(issue.path)}${issue.line ? `:${issue.line}` : ''}` : '';
result.push(`${issue.authorship === 'preexisting' ? 'Pre-existing finding\n\n' : ''}${issueBody(issue)}${location}`);
}
if (duplicateCount) result.push(`${duplicateCount} finding(s) already reported by this publisher on unchanged file content were omitted.`);
if (value.review.suggestions.length) result.push('Suggestions:\n\n' + value.review.suggestions.map(item =>
`- ${item.path ? `${markdown(item.path)}${item.line ? `:${item.line}` : ''}: ` : ''}${markdown(item.suggestion)}`).join('\n'));
for (const [field, label] of [['not_reviewed', 'Not reviewed'], ['verified', 'Verified'], ['ruled_out', 'Checked without findings']]) {
if (value.review[field].length) result.push(`${label}:\n\n${value.review[field].map(item => `- ${markdown(item)}`).join('\n')}`);
}
if (value.review.overall_assessment) result.push(markdown(value.review.overall_assessment));
result.push('Advisory review. Binary contents and arbitrary unchanged callers are outside its coverage.');
return result.filter(Boolean).join('\n\n');
}
+161
View File
@@ -0,0 +1,161 @@
import { appendFileSync, lstatSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';
import { spawnSync } from 'node:child_process';
import { fingerprint, issueBody, parseArtifact, renderSummary, repositoryPath } from './findings.mjs';
import { target, validateSnapshot } from './review.mjs';
function stop(outcome, reason) {
throw Object.assign(new Error(reason), { outcome, reason });
}
export function rightLines(patch) {
const ranges = [];
let oldRemaining = 0;
let newRemaining = 0;
let line = 0;
let hasHunk = false;
for (const entry of (patch || '').split('\n')) {
const hunk = /^@@ -\d+(?:,(\d+))? \+(\d+)(?:,(\d+))? @@/.exec(entry);
if (hunk) {
if (oldRemaining || newRemaining) return [];
oldRemaining = Number(hunk[1] ?? 1);
line = Number(hunk[2]);
newRemaining = Number(hunk[3] ?? 1);
if (![oldRemaining, line, newRemaining].every(Number.isSafeInteger)) return [];
hasHunk = true;
continue;
}
if (!hasHunk) continue;
if (entry.startsWith('\\')) continue;
if (!oldRemaining && !newRemaining && entry === '') continue;
if (entry.startsWith('+') && newRemaining > 0) {
ranges.push(line++);
newRemaining--;
} else if (entry.startsWith('-') && oldRemaining > 0) {
oldRemaining--;
} else if (entry.startsWith(' ') && oldRemaining > 0 && newRemaining > 0) {
ranges.push(line++);
oldRemaining--;
newRemaining--;
} else return [];
}
return oldRemaining || newRemaining ? [] : ranges;
}
export function planPublication(artifact, files, reviews, reviewer) {
const marker = `<!-- codewhale-action-review:v1:${fingerprint([artifact.repository, artifact.pr_number,
artifact.base_sha, artifact.head_sha, artifact.review])} -->`;
const authored = reviews.filter(item => item.user?.login === reviewer && item.state === 'COMMENTED');
if (authored.some(item => item.body?.includes(marker))) return null;
const known = new Set(authored.flatMap(item => [...(item.body || '').matchAll(/<!-- codewhale-action-finding:v1:([a-f0-9]{64}) -->/g)].map(match => match[1])));
const byPath = new Map(files.map(file => [file.filename, file]));
const normalized = value => value.trim().replace(/\s+/g, ' ').toLowerCase();
const selected = [];
for (const issue of artifact.review.issues) {
const file = byPath.get(issue.path);
const id = fingerprint([artifact.repository, artifact.pr_number, issue.path,
file?.sha || artifact.head_sha, issue.line, issue.severity, issue.authorship,
normalized(issue.title), normalized(issue.description)]);
if (!known.has(id)) selected.push({ issue, id, file });
known.add(id);
}
if (artifact.review.issues.length && selected.length === 0 && artifact.review.suggestions.length === 0) return null;
const body = [renderSummary(artifact, selected.map(item => item.issue), artifact.review.issues.length - selected.length),
marker, ...selected.map(item => `<!-- codewhale-action-finding:v1:${item.id} -->`)].join('\n\n');
const comments = selected.filter(({ issue, file }) => file && issue.line && issue.inline_anchorable !== false
&& issue.authorship !== 'preexisting' && issue.evidence?.tier !== 'suspected'
&& rightLines(file.patch).includes(issue.line)).slice(0, 50)
.map(({ issue }) => ({ path: issue.path, line: issue.line, side: 'RIGHT', body: issueBody(issue) }));
if (Buffer.byteLength(body) > 60000 || comments.some(item => Buffer.byteLength(item.body) > 60000)) {
stop('failed', 'review_exceeds_github_limit');
}
return { commit_id: artifact.head_sha, event: 'COMMENT', body, comments };
}
function command(program, args, options) {
return spawnSync(program, args, { encoding: 'utf8', maxBuffer: 16 * 1024 * 1024,
timeout: 120000, killSignal: 'SIGKILL', ...options });
}
export function publish(env = process.env, execute = command) {
const root = mkdtempSync(join(env.RUNNER_TEMP || tmpdir(), 'codewhale-publish-'));
const receiptPath = join(root, 'publication.json');
let receipt = { schema_version: 1, outcome: 'failed', reason: 'setup_failed', publication: 'not_attempted' };
try {
const event = JSON.parse(readFileSync(env.GITHUB_EVENT_PATH, 'utf8'));
const { repo, number } = target(env, event);
if (!env.GH_TOKEN) stop('configuration_missing', 'github_token_required');
const reviewer = env.CW_REVIEWER_LOGIN || 'github-actions[bot]';
if (!/^[a-zA-Z0-9][a-zA-Z0-9-]{0,99}(?:\[bot\])?$/.test(reviewer)) stop('configuration_missing', 'invalid_reviewer_login');
const file = lstatSync(env.CW_FINDINGS || '');
if (!file.isFile() || file.size > 2 * 1024 * 1024) stop('configuration_missing', 'invalid_findings_artifact');
let artifact;
try { artifact = parseArtifact(readFileSync(env.CW_FINDINGS, 'utf8')); }
catch { stop('configuration_missing', 'invalid_findings_artifact'); }
if (artifact.repository !== repo || artifact.pr_number !== number
|| artifact.run_id !== env.GITHUB_RUN_ID || artifact.run_attempt !== env.GITHUB_RUN_ATTEMPT) {
stop('configuration_missing', 'artifact_context_mismatch');
}
receipt = { ...receipt, repository: repo, pr_number: number, head_sha: artifact.head_sha,
base_sha: artifact.base_sha, findings: artifact.review.issues.length };
const childEnv = Object.fromEntries(['PATH', 'SystemRoot', 'TMPDIR', 'LANG'].filter(key => env[key]).map(key => [key, env[key]]));
Object.assign(childEnv, { HOME: root, GH_CONFIG_DIR: join(root, 'gh'), GH_TOKEN: env.GH_TOKEN,
GH_HOST: 'github.com', GH_PROMPT_DISABLED: '1', NO_COLOR: '1', CI: 'true' });
const api = (path, paginate = false, input) => {
const args = ['api', path, ...(paginate ? ['--paginate', '--slurp'] : []),
...(input ? ['--method', 'POST', '--input', '-'] : [])];
const result = execute('gh', args, { cwd: root, env: childEnv, ...(input ? { input: JSON.stringify(input) } : {}) });
if (result.status !== 0 || result.error) stop(input ? 'publication_uncertain' : 'failed',
input ? 'inspect_github_before_retry' : 'github_read_failed');
let data;
try { data = JSON.parse(result.stdout); }
catch { stop(input ? 'publication_uncertain' : 'failed', 'invalid_github_response'); }
if (paginate) {
if (!Array.isArray(data) || data.some(page => !Array.isArray(page))) stop('failed', 'invalid_github_response');
return data.flat();
}
return data;
};
const snapshot = () => {
const pr = api(`repos/${repo}/pulls/${number}`);
validateSnapshot(pr, repo, event);
if (pr.head.sha !== artifact.head_sha || pr.base.sha !== artifact.base_sha) stop('superseded', 'revision_changed');
return pr;
};
const pr = snapshot();
const files = api(`repos/${repo}/pulls/${number}/files?per_page=100`, true);
if (!Number.isSafeInteger(pr.changed_files) || files.length !== pr.changed_files
|| new Set(files.map(item => item.filename)).size !== files.length
|| files.some(item => !repositoryPath(item.filename) || !/^[a-f0-9]{40}(?:[a-f0-9]{24})?$/i.test(item.sha || '')
|| (item.patch != null && (typeof item.patch !== 'string' || Buffer.byteLength(item.patch) > 8 * 1024 * 1024)))) {
stop('failed', 'incomplete_github_file_list');
}
const reviews = api(`repos/${repo}/pulls/${number}/reviews?per_page=100`, true);
const plan = planPublication(artifact, files, reviews, reviewer);
snapshot();
if (!plan) receipt = { ...receipt, outcome: 'already_published', reason: '', publication: 'deduplicated' };
else {
const posted = api(`repos/${repo}/pulls/${number}/reviews`, false, plan);
if (!Number.isSafeInteger(posted.id) || posted.id <= 0 || posted.commit_id !== artifact.head_sha
|| posted.state !== 'COMMENTED' || posted.user?.login !== reviewer) {
stop('publication_uncertain', 'inspect_github_before_retry');
}
receipt = { ...receipt, outcome: 'published', reason: '', publication: 'posted', review_id: posted.id,
inline_comments: plan.comments.length };
}
} catch (error) {
receipt = { ...receipt, outcome: error.outcome || 'failed', reason: error.reason || 'setup_failed',
...(error.outcome === 'publication_uncertain' ? { publication: 'uncertain' } : {}) };
}
writeFileSync(receiptPath, `${JSON.stringify(receipt, null, 2)}\n`, { mode: 0o600 });
if (env.GITHUB_OUTPUT) appendFileSync(env.GITHUB_OUTPUT, `outcome=${receipt.outcome}\nreceipt=${receiptPath}\n`);
if (env.GITHUB_STEP_SUMMARY) appendFileSync(env.GITHUB_STEP_SUMMARY,
`### Codewhale publication\n\nOutcome: **${receipt.outcome}**. ${receipt.reason ? `Reason: \`${receipt.reason}\`.` : ''}\n`);
const ok = ['published', 'already_published', 'superseded', 'not_eligible'].includes(receipt.outcome);
console.log(`${ok ? '::notice::' : '::error::'}Codewhale publication: ${receipt.outcome} (${receipt.reason || 'complete'}).`);
return { exitCode: ok ? 0 : 1, receipt, receiptPath };
}
if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) process.exitCode = publish().exitCode;
+33
View File
@@ -0,0 +1,33 @@
name: Publish Codewhale review
description: Validate and publish a completed Codewhale findings artifact in a separate job without model credentials.
inputs:
findings:
description: Path to findings.json from this workflow run and attempt.
required: true
github-token:
description: GitHub token with contents read and pull-requests write.
default: ${{ github.token }}
pr-number:
description: PR number for workflow_dispatch; pull_request events use the event snapshot.
reviewer-login:
description: Exact GitHub login used by github-token; only this author's Codewhale markers deduplicate reviews.
default: github-actions[bot]
outputs:
outcome:
description: published, already_published, superseded, not_eligible, configuration_missing, failed, or publication_uncertain.
value: ${{ steps.publish.outputs.outcome }}
receipt:
description: Absolute path to the sanitized JSON publication receipt.
value: ${{ steps.publish.outputs.receipt }}
runs:
using: composite
steps:
- id: publish
shell: bash
env:
CW_ACTION_PATH: ${{ github.action_path }}
CW_FINDINGS: ${{ inputs.findings }}
CW_PR_NUMBER: ${{ inputs.pr-number }}
CW_REVIEWER_LOGIN: ${{ inputs.reviewer-login }}
GH_TOKEN: ${{ inputs.github-token }}
run: node "$CW_ACTION_PATH/../publish.mjs"
+15 -3
View File
@@ -2,12 +2,13 @@
// model routing, anchor validation and the final pre-publication head check.
// No repository checkout, project configuration, hooks, or PR code is executed.
// Known limits: review only (mentions use the hosted App); no dollar-budget
// guarantee, automatic retry, or cross-run publication deduplication.
// guarantee or automatic retry. Legacy direct publication has no cross-run deduplication.
import { appendFileSync, mkdirSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';
import { spawnSync } from 'node:child_process';
import { findingsArtifact } from './findings.mjs';
const SHA = /^[a-f0-9]{40}(?:[a-f0-9]{24})?$/i;
const REPO = /^[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+$/;
@@ -124,6 +125,7 @@ function command(program, args, options) {
export function run(env = process.env, execute = command) {
const root = mkdtempSync(join(env.RUNNER_TEMP || tmpdir(), 'codewhale-review-'));
const receiptPath = join(root, 'receipt.json');
let findingsPath = '';
let receipt = { schema_version: 1, outcome: 'failed', reason: 'setup_failed', publication: 'not_attempted' };
let prUrl = '';
try {
@@ -178,19 +180,29 @@ export function run(env = process.env, execute = command) {
if (config.post) args.push('--post');
const result = execute(cli, args, { cwd: repoDir, env: childEnv, timeout: config.timeout });
receipt = { ...receipt, ...classify(result, pr.head.sha, config.post, pr.base.sha) };
if (!config.post && ['reviewed_clean', 'reviewed_with_findings'].includes(receipt.outcome)) {
const data = JSON.parse(result.stdout);
if (data.pr?.number !== number || data.pr?.url !== prUrl) stop('failed', 'review_target_mismatch');
let artifact;
try { artifact = findingsArtifact(data, receipt, env); }
catch { stop('failed', 'invalid_review_findings'); }
const path = join(root, 'findings.json');
writeFileSync(path, `${JSON.stringify(artifact, null, 2)}\n`, { mode: 0o600 });
findingsPath = path;
}
} catch (error) {
receipt = { ...receipt, outcome: error instanceof Stop ? error.outcome : 'failed',
reason: error instanceof Stop ? error.reason : 'setup_failed' };
}
writeFileSync(receiptPath, `${JSON.stringify(receipt, null, 2)}\n`, { mode: 0o600 });
if (env.GITHUB_OUTPUT) appendFileSync(env.GITHUB_OUTPUT, `outcome=${receipt.outcome}\nreceipt=${receiptPath}\npr-url=${prUrl}\n`);
if (env.GITHUB_OUTPUT) appendFileSync(env.GITHUB_OUTPUT, `outcome=${receipt.outcome}\nreceipt=${receiptPath}\nfindings=${findingsPath}\npr-url=${prUrl}\n`);
if (env.GITHUB_STEP_SUMMARY) appendFileSync(env.GITHUB_STEP_SUMMARY,
`### Codewhale review\n\nOutcome: **${receipt.outcome}**. ${receipt.reason ? `Reason: \`${receipt.reason}\`.` : ''}\n\n` +
'Only `reviewed_clean` and `reviewed_with_findings` confirm complete model review. Findings are advisory.\n');
const ok = ['reviewed_clean', 'reviewed_with_findings', 'not_eligible', 'superseded'].includes(receipt.outcome);
// Constant vocabulary only; model/provider/repository text is never logged.
console.log(`${ok ? '::notice::' : '::error::'}Codewhale review: ${receipt.outcome} (${receipt.reason || 'complete'}).`);
return { exitCode: ok ? 0 : 1, receipt, receiptPath };
return { exitCode: ok ? 0 : 1, receipt, receiptPath, findingsPath };
}
if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) process.exitCode = run().exitCode;
+8 -2
View File
@@ -12,9 +12,10 @@ const pr = { state: 'open', draft: false, base: { sha: BASE, repo: { full_name:
const event = { action: 'opened', number: 12, pull_request: pr };
const defaults = { GITHUB_REPOSITORY: 'owner/repo', GITHUB_EVENT_NAME: 'pull_request',
CW_VERSION: 'v0.10.0', CW_PROVIDER: 'codewhale', CW_MODEL: 'provider/model',
CW_ACTION_PATH: '/trusted/action', GH_TOKEN: 'github-secret', CODEWHALE_API_KEY: 'account-secret' };
CW_ACTION_PATH: '/trusted/action', GH_TOKEN: 'github-secret', CODEWHALE_API_KEY: 'account-secret',
GITHUB_RUN_ID: '123', GITHUB_RUN_ATTEMPT: '1' };
const completion = (override = {}) => ({ mode: 'review', success: true, complete: true,
pr: { head_sha: HEAD }, review: { issues: [] }, review_passes: 1, publication: 'posted',
pr: { head_sha: HEAD, number: 12, url: 'https://github.com/owner/repo/pull/12' }, review: { summary: 'No findings', issues: [], suggestions: [], overall_assessment: 'No findings' }, review_passes: 1, publication: 'posted',
receipt: { coverage: { manifest: { head_sha: HEAD, base_sha: BASE, skipped_files: [] }, completed_passes: [{}] } },
usage: { input_tokens: 10, output_tokens: 20 }, ...override });
const output = data => ({ status: 0, stdout: JSON.stringify(data), stderr: '' });
@@ -144,4 +145,9 @@ test('findings are advisory and preview has no publication claim', t => {
const preview = fixture(t, { env: { CW_POST: 'false' }, intercept: (_p, args) => args.includes('review') && output(completion({ publication: 'not_requested' })) });
assert.equal(preview.exitCode, 0);
assert.equal(preview.calls.find(c => c.args.includes('review')).args.includes('--post'), false);
const artifact = JSON.parse(readFileSync(preview.findingsPath, 'utf8'));
assert.equal(artifact.publication, 'not_requested');
assert.equal(artifact.head_sha, HEAD);
assert.deepEqual(artifact.review.issues, []);
assert.match(preview.outputs, /findings=.*findings\.json/);
});
+40 -1
View File
@@ -2,7 +2,7 @@
"format": "codewhale.features.v1",
"source": {
"path": "docs/features.toml",
"sha256": "42d526ee838721c73b17e03178a6c61f16fd9b7748553bc12ffa2679451aae2c"
"sha256": "38885b15b57125002edb13f12c4f0a13772ddce9cf63c2d68c4b1001cd63b3dc"
},
"features": [
{
@@ -1087,6 +1087,45 @@
"flag": null,
"docs": null,
"registryLine": 804
},
{
"id": "github-action-review",
"name": "GitHub Action review",
"summary": "Review pull requests with the Engine and publish validated findings from a separate trusted job.",
"status": "experimental",
"surfaces": {
"tui": "preview"
},
"since": "unreleased",
"flag": null,
"docs": "docs/GITHUB_ACTION.md",
"registryLine": 813
},
{
"id": "telegram-agent",
"name": "Telegram agent bridge",
"summary": "Use an Engine from Telegram with durable prompt recovery, human-owned approvals and retained replies.",
"status": "experimental",
"surfaces": {
"api": "preview"
},
"since": "unreleased",
"flag": null,
"docs": "integrations/telegram-bridge/README.md",
"registryLine": 823
},
{
"id": "dingtalk-agent",
"name": "DingTalk agent bridge",
"summary": "Connect an internal DingTalk app to the same Engine with scoped identity and durable message recovery.",
"status": "experimental",
"surfaces": {
"api": "preview"
},
"since": "unreleased",
"flag": null,
"docs": "integrations/dingtalk-bridge/README.md",
"registryLine": 833
}
]
}