feat(knowledge): add central ingest workflows

.github/workflows/deploy-repo-gate.yml

@@ -66,6 +66,9 @@ jobs:
       - name: Validate release queue structure
         run: uvx --with pyyaml python scripts/promoter/validate_queue.py
 
+      - name: Validate knowledge publish records
+        run: uvx --with pyyaml --with jsonschema python -m scripts.knowledge.validate_state --deploy-root . --env dev
+
       - name: Evidence validate + collect (dry run)
         run: |
           uvx --with pyyaml --with jsonschema python scripts/evidence/validate.py

.github/workflows/knowledge-ingest.yml

@@ -0,0 +1,119 @@
+name: knowledge-ingest
+
+on:
+  repository_dispatch:
+    types:
+      - knowledge_artifact_ready
+  workflow_dispatch:
+    inputs:
+      source_repo:
+        description: "source repository (owner/repo)"
+        required: true
+        type: string
+      source_run_id:
+        description: "source workflow run id"
+        required: true
+        type: string
+      artifact_name:
+        description: "artifact name to ingest"
+        required: true
+        type: string
+      target_env:
+        description: "target environment"
+        required: true
+        default: dev
+        type: choice
+        options:
+          - dev
+      skip_dify:
+        description: "skip live Dify publication"
+        required: true
+        default: "false"
+        type: choice
+        options:
+          - "false"
+          - "true"
+
+permissions:
+  contents: write
+
+concurrency:
+  group: knowledge-ingest-${{ github.event_name }}-${{ github.event_name == 'repository_dispatch' && github.event.client_payload.source_repo || inputs.source_repo }}-${{ github.event_name == 'repository_dispatch' && github.event.client_payload.source_run_id || inputs.source_run_id }}
+  cancel-in-progress: false
+
+jobs:
+  ingest:
+    runs-on: ubuntu-latest
+    env:
+      KNOWLEDGE_DIFY_BASE_URL: ${{ secrets.KNOWLEDGE_DIFY_BASE_URL }}
+      KNOWLEDGE_DIFY_API_KEY: ${{ secrets.KNOWLEDGE_DIFY_API_KEY }}
+      KNOWLEDGE_GITHUB_TOKEN: ${{ secrets.DEPLOY_REPO_TOKEN || github.token }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Setup uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Resolve inputs
+        id: prep
+        shell: bash
+        run: |
+          set -euo pipefail
+          if [ "${GITHUB_EVENT_NAME}" = "repository_dispatch" ]; then
+            echo "source_repo=${{ github.event.client_payload.source_repo }}" >> "${GITHUB_OUTPUT}"
+            echo "source_run_id=${{ github.event.client_payload.source_run_id }}" >> "${GITHUB_OUTPUT}"
+            echo "artifact_name=${{ github.event.client_payload.artifact_name }}" >> "${GITHUB_OUTPUT}"
+            echo "target_env=${{ github.event.client_payload.target_env || 'dev' }}" >> "${GITHUB_OUTPUT}"
+            echo "skip_dify=${{ github.event.client_payload.skip_dify || 'false' }}" >> "${GITHUB_OUTPUT}"
+            echo "source_sha=${{ github.event.client_payload.source_sha || '' }}" >> "${GITHUB_OUTPUT}"
+          else
+            echo "source_repo=${{ inputs.source_repo }}" >> "${GITHUB_OUTPUT}"
+            echo "source_run_id=${{ inputs.source_run_id }}" >> "${GITHUB_OUTPUT}"
+            echo "artifact_name=${{ inputs.artifact_name }}" >> "${GITHUB_OUTPUT}"
+            echo "target_env=${{ inputs.target_env }}" >> "${GITHUB_OUTPUT}"
+            echo "skip_dify=${{ inputs.skip_dify }}" >> "${GITHUB_OUTPUT}"
+            echo "source_sha=" >> "${GITHUB_OUTPUT}"
+          fi
+
+      - name: Ingest artifact
+        run: |
+          uvx --with pyyaml --with jsonschema python -m scripts.knowledge.ingest_artifact \
+            --deploy-root . \
+            --env "${{ steps.prep.outputs.target_env }}" \
+            --source-repo "${{ steps.prep.outputs.source_repo }}" \
+            --source-run-id "${{ steps.prep.outputs.source_run_id }}" \
+            --artifact-name "${{ steps.prep.outputs.artifact_name }}" \
+            --github-token "${KNOWLEDGE_GITHUB_TOKEN}" \
+            --dify-base-url "${KNOWLEDGE_DIFY_BASE_URL}" \
+            --dify-api-key "${KNOWLEDGE_DIFY_API_KEY}" \
+            ${{ steps.prep.outputs.skip_dify == 'true' && '--skip-dify' || '' }} \
+            --summary-json /tmp/knowledge-ingest-summary.json
+          cat /tmp/knowledge-ingest-summary.json
+
+      - name: Validate committed knowledge state
+        run: |
+          uvx --with pyyaml --with jsonschema python -m scripts.knowledge.validate_state \
+            --deploy-root . \
+            --env "${{ steps.prep.outputs.target_env }}"
+
+      - name: Commit knowledge state
+        shell: bash
+        run: |
+          set -euo pipefail
+          if git diff --quiet -- knowledge; then
+            echo "No knowledge changes to commit"
+            exit 0
+          fi
+          git config user.name "knowledge-publisher[bot]"
+          git config user.email "knowledge-publisher[bot]@users.noreply.github.com"
+          git add knowledge
+          git commit -m "feat(knowledge): ingest artifact from ${{ steps.prep.outputs.source_repo }}@${{ steps.prep.outputs.source_sha || 'workflow' }}"
+          git push

.github/workflows/knowledge-reconcile.yml

@@ -0,0 +1,84 @@
+name: knowledge-reconcile
+
+on:
+  workflow_dispatch:
+    inputs:
+      target_env:
+        description: "target environment"
+        required: true
+        default: dev
+        type: choice
+        options:
+          - dev
+      source_repo:
+        description: "optional source repo filter"
+        required: false
+        type: string
+      document_id:
+        description: "optional document id filter"
+        required: false
+        type: string
+      skip_dify:
+        description: "skip live Dify publication"
+        required: true
+        default: "false"
+        type: choice
+        options:
+          - "false"
+          - "true"
+
+permissions:
+  contents: write
+
+jobs:
+  reconcile:
+    runs-on: ubuntu-latest
+    env:
+      KNOWLEDGE_DIFY_BASE_URL: ${{ secrets.KNOWLEDGE_DIFY_BASE_URL }}
+      KNOWLEDGE_DIFY_API_KEY: ${{ secrets.KNOWLEDGE_DIFY_API_KEY }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Setup uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Reconcile publish records
+        run: |
+          uvx --with pyyaml --with jsonschema python -m scripts.knowledge.reconcile_records \
+            --deploy-root . \
+            --env "${{ inputs.target_env }}" \
+            ${{ inputs.source_repo != '' && format('--source-repo "{0}"', inputs.source_repo) || '' }} \
+            ${{ inputs.document_id != '' && format('--document-id "{0}"', inputs.document_id) || '' }} \
+            --dify-base-url "${KNOWLEDGE_DIFY_BASE_URL}" \
+            --dify-api-key "${KNOWLEDGE_DIFY_API_KEY}" \
+            ${{ inputs.skip_dify == 'true' && '--skip-dify' || '' }} \
+            --summary-json /tmp/knowledge-reconcile-summary.json
+          cat /tmp/knowledge-reconcile-summary.json
+
+      - name: Validate committed knowledge state
+        run: |
+          uvx --with pyyaml --with jsonschema python -m scripts.knowledge.validate_state \
+            --deploy-root . \
+            --env "${{ inputs.target_env }}"
+
+      - name: Commit reconciled state
+        shell: bash
+        run: |
+          set -euo pipefail
+          if git diff --quiet -- knowledge; then
+            echo "No knowledge changes to commit"
+            exit 0
+          fi
+          git config user.name "knowledge-publisher[bot]"
+          git config user.email "knowledge-publisher[bot]@users.noreply.github.com"
+          git add knowledge
+          git commit -m "feat(knowledge): reconcile published records"
+          git push

.github/workflows/knowledge-unpublish.yml

@@ -0,0 +1,90 @@
+name: knowledge-unpublish
+
+on:
+  workflow_dispatch:
+    inputs:
+      target_env:
+        description: "target environment"
+        required: true
+        default: dev
+        type: choice
+        options:
+          - dev
+      source_repo:
+        description: "source repository (owner/repo)"
+        required: true
+        type: string
+      document_id:
+        description: "document id to unpublish"
+        required: true
+        type: string
+      reason:
+        description: "unpublish reason"
+        required: true
+        default: manual_unpublish
+        type: string
+      skip_dify:
+        description: "skip live Dify publication"
+        required: true
+        default: "false"
+        type: choice
+        options:
+          - "false"
+          - "true"
+
+permissions:
+  contents: write
+
+jobs:
+  unpublish:
+    runs-on: ubuntu-latest
+    env:
+      KNOWLEDGE_DIFY_BASE_URL: ${{ secrets.KNOWLEDGE_DIFY_BASE_URL }}
+      KNOWLEDGE_DIFY_API_KEY: ${{ secrets.KNOWLEDGE_DIFY_API_KEY }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Setup uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Unpublish record
+        run: |
+          uvx --with pyyaml --with jsonschema python -m scripts.knowledge.unpublish_records \
+            --deploy-root . \
+            --env "${{ inputs.target_env }}" \
+            --source-repo "${{ inputs.source_repo }}" \
+            --document-id "${{ inputs.document_id }}" \
+            --reason "${{ inputs.reason }}" \
+            --dify-base-url "${KNOWLEDGE_DIFY_BASE_URL}" \
+            --dify-api-key "${KNOWLEDGE_DIFY_API_KEY}" \
+            ${{ inputs.skip_dify == 'true' && '--skip-dify' || '' }} \
+            --summary-json /tmp/knowledge-unpublish-summary.json
+          cat /tmp/knowledge-unpublish-summary.json
+
+      - name: Validate committed knowledge state
+        run: |
+          uvx --with pyyaml --with jsonschema python -m scripts.knowledge.validate_state \
+            --deploy-root . \
+            --env "${{ inputs.target_env }}"
+
+      - name: Commit unpublished state
+        shell: bash
+        run: |
+          set -euo pipefail
+          if git diff --quiet -- knowledge; then
+            echo "No knowledge changes to commit"
+            exit 0
+          fi
+          git config user.name "knowledge-publisher[bot]"
+          git config user.email "knowledge-publisher[bot]@users.noreply.github.com"
+          git add knowledge
+          git commit -m "feat(knowledge): unpublish ${{ inputs.document_id }}"
+          git push

docs/platform/knowledge-pipeline.md

@@ -2,27 +2,37 @@
 
 ## Positioning
 
-The raw source of truth stays in Git-managed repositories and approved source systems.
-Dify only consumes processed outputs.
+The raw source of truth stays in Git-managed business repositories and approved source systems.
+Dify only consumes centrally processed outputs.
 
-The baseline Git-managed knowledge repository is `ljwx-knowledge` with:
+The current MVP is a two-layer pipeline:
 
-- `content/public/**`
-- `content/internal/**`
-- `content/customers/{customer_id}/**`
+1. business repositories export standardized knowledge artifacts
+2. `ljwx-deploy` ingests those artifacts, applies centralized policy, and publishes to Dify dev
 
 ## Pipeline
 
-1. collect
+### Layer 1: Business Repo Export
+
+Each business repo only owns:
+
+1. declare included markdown sources in `knowledge.sources.yaml`
+2. run light validation
+3. export a portable artifact (`manifest.json` + `documents/*.md`)
+4. dispatch the artifact to `ljwx-deploy`
+
+### Layer 2: Central Ingest
+
+`ljwx-deploy` owns:
+
+1. artifact validation
 2. normalize
 3. classify
 4. deduplicate
 5. chunk
-6. tag
-7. apply ACL
-8. review
-9. publish
-10. invalidate or unpublish
+6. apply ACL
+7. publish / invalidate / republish
+8. write publish records and event history in Git
 
 ## Publish Targets
 
@@ -30,6 +40,13 @@ The baseline Git-managed knowledge repository is `ljwx-knowledge` with:
 - internal retrieval collections for platform and support use cases
 - customer-scoped retrieval collections when tenant isolation is required
 
+MVP scope:
+
+1. markdown only
+2. `public` / `internal` visibility only
+3. dev Dify dataset publication first
+4. retrieval targets stay declared in config but are not yet executed by the MVP workflow
+
 ## Invalidate Triggers
 
 Publishing is not append-only. A document or chunk must be invalidated when:
@@ -44,26 +61,29 @@ The target configuration lives in `platform/knowledge/publish-targets.yaml`.
 
 ## Directory and ACL Contract
 
-- `platform/knowledge/source-registry.yaml` defines which repositories and paths are collectible.
+- `platform/knowledge/source-registry.yaml` defines which repositories can export artifacts and which source IDs are centrally allowed.
 - `platform/knowledge/taxonomy.yaml` defines the path-to-classification boundary.
 - `platform/knowledge/acl-policies.yaml` defines visibility-specific `acl_expression_template` and allowed publish targets.
-- `platform/contracts/knowledge-document.schema.yaml` and `platform/contracts/knowledge-chunk.schema.yaml` define the minimum manifest shape for normalize/classify/chunk outputs.
+- `platform/contracts/knowledge-source-config.schema.yaml` defines business repo export config.
+- `platform/contracts/knowledge-artifact.schema.yaml` defines the exported artifact manifest.
+- `platform/contracts/knowledge-document.schema.yaml` and `platform/contracts/knowledge-chunk.schema.yaml` define the minimum normalize/classify/chunk outputs.
+- `platform/contracts/knowledge-publish-record.schema.yaml` defines the committed central publication record.
 
 ## Minimum Execution Flow
 
 The current MVP path is:
 
-1. ingest markdown sources from `ljwx-knowledge`
-2. emit normalized document manifests
-3. enrich `doc_type`, `product_area`, and `acl_expression`
-4. chunk approved documents
-5. publish to local dev datasets
-6. invalidate or unpublish by `document_id` and republish when a rollback is needed
+1. business repo exports a markdown artifact from `knowledge.sources.yaml`
+2. `knowledge-ingest` validates and snapshots the artifact in `ljwx-deploy/knowledge/artifacts/dev/`
+3. central scripts emit normalized/classified/chunk manifests
+4. central publish resolves dataset mapping from `platform/knowledge/publish-targets.yaml`
+5. central workflow writes publish records to `knowledge/publish-records/dev/`
+6. `knowledge-unpublish` and `knowledge-reconcile` provide the minimal invalidate / republish loop
 
 ## Audit And Metrics
 
-The current MVP also emits:
+The current MVP emits:
 
-1. `state/<env>/events.jsonl` for lifecycle events
-2. `state/<env>/audit-events.jsonl` for cross-component audit evidence
-3. `state/<env>/metrics.json` for the latest publish or rollback counters
+1. `knowledge/events/<env>/events.jsonl` for append-only lifecycle events
+2. `knowledge/publish-records/<env>/**/*.json` for the latest central publication state
+3. `knowledge/artifacts/<env>/**` for replayable raw artifacts and derived manifests

platform/contracts/knowledge-artifact.schema.yaml

@@ -0,0 +1,166 @@
+$schema: https://json-schema.org/draft/2020-12/schema
+$id: https://lingjingwanxiang.cn/schemas/knowledge-artifact.schema.yaml
+title: KnowledgeArtifact
+type: object
+additionalProperties: false
+required:
+  - schema_version
+  - artifact_id
+  - generated_at
+  - export_mode
+  - source
+  - documents
+properties:
+  schema_version:
+    type: string
+    minLength: 1
+  artifact_id:
+    type: string
+    minLength: 1
+  generated_at:
+    type: string
+    format: date-time
+  export_mode:
+    type: string
+    enum:
+      - changed
+      - full
+  source:
+    type: object
+    additionalProperties: false
+    required:
+      - repo
+      - service
+      - commit
+      - ref
+      - workflow
+      - run_id
+      - run_attempt
+    properties:
+      repo:
+        type: string
+        minLength: 1
+      service:
+        type: string
+        minLength: 1
+      commit:
+        type: string
+        minLength: 1
+      ref:
+        type: string
+        minLength: 1
+      workflow:
+        type: string
+        minLength: 1
+      run_id:
+        type: integer
+        minimum: 1
+      run_attempt:
+        type: integer
+        minimum: 1
+  documents:
+    type: array
+    items:
+      type: object
+      additionalProperties: false
+      required:
+        - op
+        - document_id
+        - source_id
+        - source_path
+        - source_ref
+      properties:
+        op:
+          type: string
+          enum:
+            - upsert
+            - invalidate
+        document_id:
+          type: string
+          minLength: 1
+        source_id:
+          type: string
+          minLength: 1
+        source_path:
+          type: string
+          minLength: 1
+        source_ref:
+          type: string
+          minLength: 1
+        content_path:
+          type:
+            - string
+            - "null"
+        title:
+          type:
+            - string
+            - "null"
+        visibility:
+          anyOf:
+            - type: string
+              enum:
+                - public
+                - internal
+            - type: "null"
+        source_type:
+          anyOf:
+            - type: string
+              enum:
+                - repo_doc
+                - faq
+                - delivery_doc
+                - deployment_doc
+                - support_record
+                - partner_reference
+            - type: "null"
+        doc_type:
+          type:
+            - string
+            - "null"
+        product_area:
+          type:
+            - string
+            - "null"
+        owner:
+          type:
+            - string
+            - "null"
+        review_status:
+          anyOf:
+            - type: string
+              enum:
+                - draft
+                - reviewing
+                - approved
+                - rejected
+            - type: "null"
+        version:
+          type:
+            - string
+            - "null"
+        checksum:
+          type:
+            - string
+            - "null"
+        tags:
+          type: array
+          items:
+            type: string
+        approved_by:
+          type:
+            - string
+            - "null"
+        effective_from:
+          type:
+            - string
+            - "null"
+          format: date-time
+        expires_at:
+          type:
+            - string
+            - "null"
+          format: date-time
+        reason:
+          type:
+            - string
+            - "null"

platform/contracts/knowledge-publish-record.schema.yaml

@@ -0,0 +1,189 @@
+$schema: https://json-schema.org/draft/2020-12/schema
+$id: https://lingjingwanxiang.cn/schemas/knowledge-publish-record.schema.yaml
+title: KnowledgePublishRecord
+type: object
+additionalProperties: false
+required:
+  - schema_version
+  - document_id
+  - source_repo
+  - source_id
+  - source_path
+  - visibility
+  - checksum
+  - status
+  - artifact_ref
+  - content_ref
+  - normalized_ref
+  - classified_ref
+  - chunk_ref
+  - latest_action
+  - updated_at
+  - targets
+properties:
+  schema_version:
+    type: string
+    minLength: 1
+  document_id:
+    type: string
+    minLength: 1
+  source_repo:
+    type: string
+    minLength: 1
+  source_id:
+    type: string
+    minLength: 1
+  source_path:
+    type: string
+    minLength: 1
+  source_ref:
+    type:
+      - string
+      - "null"
+  visibility:
+    type: string
+    enum:
+      - public
+      - internal
+  checksum:
+    type:
+      - string
+      - "null"
+  title:
+    type:
+      - string
+      - "null"
+  source_type:
+    type:
+      - string
+      - "null"
+  doc_type:
+    type:
+      - string
+      - "null"
+  product_area:
+    type:
+      - string
+      - "null"
+  owner:
+    type:
+      - string
+      - "null"
+  review_status:
+    anyOf:
+      - type: string
+        enum:
+          - draft
+          - reviewing
+          - approved
+          - rejected
+      - type: "null"
+  acl_expression:
+    type:
+      - string
+      - "null"
+  version:
+    type:
+      - string
+      - "null"
+  tags:
+    type: array
+    items:
+      type: string
+  status:
+    type: string
+    enum:
+      - published
+      - invalidated
+      - unpublished
+      - deduplicated
+  duplicate_of:
+    type:
+      - string
+      - "null"
+  artifact_ref:
+    type: string
+    minLength: 1
+  content_ref:
+    type:
+      - string
+      - "null"
+  normalized_ref:
+    type: string
+    minLength: 1
+  classified_ref:
+    type: string
+    minLength: 1
+  chunk_ref:
+    type: string
+    minLength: 1
+  chunk_count:
+    type: integer
+    minimum: 0
+  latest_action:
+    type: string
+    enum:
+      - publish
+      - invalidate
+      - unpublish
+      - republish
+  latest_reason:
+    type:
+      - string
+      - "null"
+  published_at:
+    type:
+      - string
+      - "null"
+    format: date-time
+  updated_at:
+    type: string
+    format: date-time
+  targets:
+    type: array
+    items:
+      type: object
+      additionalProperties: false
+      required:
+        - target_id
+        - target_type
+        - dataset_name
+        - dataset_id
+        - document_name
+        - remote_document_id
+        - status
+      properties:
+        target_id:
+          type: string
+          minLength: 1
+        target_type:
+          type: string
+          minLength: 1
+        dataset_name:
+          type:
+            - string
+            - "null"
+        dataset_id:
+          type:
+            - string
+            - "null"
+        document_name:
+          type:
+            - string
+            - "null"
+        remote_document_id:
+          type:
+            - string
+            - "null"
+        status:
+          type: string
+          enum:
+            - published
+            - invalidated
+            - unpublished
+            - skipped
+        updated_at:
+          type:
+            - string
+            - "null"
+          format: date-time

platform/contracts/knowledge-source-config.schema.yaml

@@ -0,0 +1,101 @@
+$schema: https://json-schema.org/draft/2020-12/schema
+$id: https://lingjingwanxiang.cn/schemas/knowledge-source-config.schema.yaml
+title: KnowledgeSourceConfig
+type: object
+additionalProperties: false
+required:
+  - version
+  - service
+  - sources
+properties:
+  version:
+    type: string
+    minLength: 1
+  service:
+    type: string
+    minLength: 1
+  sources:
+    type: array
+    minItems: 1
+    items:
+      type: object
+      additionalProperties: false
+      required:
+        - id
+        - include
+        - visibility
+        - source_type
+        - doc_type
+        - product_area
+        - owner
+        - review_status
+      properties:
+        id:
+          type: string
+          minLength: 1
+        include:
+          type: array
+          minItems: 1
+          items:
+            type: string
+            minLength: 1
+        exclude:
+          type: array
+          items:
+            type: string
+            minLength: 1
+        visibility:
+          type: string
+          enum:
+            - public
+            - internal
+        source_type:
+          type: string
+          enum:
+            - repo_doc
+            - faq
+            - delivery_doc
+            - deployment_doc
+            - support_record
+            - partner_reference
+        doc_type:
+          type: string
+          minLength: 1
+        product_area:
+          type: string
+          minLength: 1
+        owner:
+          type: string
+          minLength: 1
+        review_status:
+          type: string
+          enum:
+            - draft
+            - reviewing
+            - approved
+            - rejected
+        tags:
+          type: array
+          items:
+            type: string
+        title_strategy:
+          type: string
+          enum:
+            - first_heading
+            - filename
+        title_prefix:
+          type: string
+        approved_by:
+          type:
+            - string
+            - "null"
+        effective_from:
+          type:
+            - string
+            - "null"
+          format: date-time
+        expires_at:
+          type:
+            - string
+            - "null"
+          format: date-time

platform/knowledge/publish-targets.yaml

@@ -1,4 +1,4 @@
-version: 0.1.0
+version: 0.2.0
 
 targets:
   dify-public:

platform/knowledge/source-registry.yaml

@@ -1,4 +1,4 @@
-version: 0.1.0
+version: 0.2.0
 
 sources:
   - id: ljwx-knowledge-public
@@ -46,3 +46,13 @@ sources:
       - content/faq/**
     visibility: public
     owner: team-product
+
+artifact_sources:
+  - repo: BrunoGaoSZ/ljwx-bookstore
+    service: ljwx-bookstore
+    source_ids:
+      - bookstore-public-readme
+      - bookstore-internal-ai-plan
+    allowed_visibilities:
+      - public
+      - internal

release/platform-version.yaml

@@ -52,11 +52,15 @@ config_versions:
   service_capability_map: platform/contracts/service-capability-map.yaml@0.2.1
   capability_gateway_request: platform/contracts/capability-gateway-request.schema.json@0.1.0
   audit_event_schema: platform/contracts/audit-event.schema.json@0.1.0
-  knowledge_source_registry: platform/knowledge/source-registry.yaml@0.1.0
+  knowledge_source_registry: platform/knowledge/source-registry.yaml@0.2.0
+  knowledge_publish_targets: platform/knowledge/publish-targets.yaml@0.2.0
   knowledge_taxonomy: platform/knowledge/taxonomy.yaml@0.1.0
   knowledge_acl_policies: platform/knowledge/acl-policies.yaml@0.1.0
   knowledge_document_schema: platform/contracts/knowledge-document.schema.yaml@0.1.1
   knowledge_chunk_schema: platform/contracts/knowledge-chunk.schema.yaml@0.1.1
+  knowledge_source_config_schema: platform/contracts/knowledge-source-config.schema.yaml@0.1.0
+  knowledge_artifact_schema: platform/contracts/knowledge-artifact.schema.yaml@0.1.0
+  knowledge_publish_record_schema: platform/contracts/knowledge-publish-record.schema.yaml@0.1.0
   workflow_templates_repo: BrunoGaoSZ/ljwx-workflow-templates@406980fc99f7
   workflow_build_template: BrunoGaoSZ/ljwx-workflow-templates/.github/workflows/build-ghcr.yml@406980fc99f7
   workflow_enqueue_template: BrunoGaoSZ/ljwx-workflow-templates/.github/workflows/enqueue-release.yml@406980fc99f7

scripts/ci/run_checks.sh

@@ -5,6 +5,7 @@ uvx --with pyyaml --with jsonschema python scripts/evidence/validate.py
 uvx --with pyyaml --with jsonschema python scripts/evidence/collect.py --out evidence/index.json
 uvx --with pyyaml --with jsonschema python scripts/platform/validate_router_contracts.py
 uvx --with pyyaml --with jsonschema python scripts/platform/validate_capability_contracts.py
+uvx --with pyyaml --with jsonschema python -m scripts.knowledge.validate_state --deploy-root . --env dev
 uvx --with pyyaml python scripts/promoter/validate_queue.py
 uvx --with pyyaml python scripts/promoter/queue_metrics.py --queue release/queue.yaml --out /tmp/queue-health.json
 bash scripts/promoter/promote.sh --dry-run

scripts/knowledge/__init__.py

@@ -0,0 +1 @@
+"""Knowledge supply chain scripts."""

scripts/knowledge/common.py

@@ -0,0 +1,681 @@
+#!/usr/bin/env python3
+"""Shared helpers for the knowledge supply chain MVP."""
+
+from __future__ import annotations
+
+import fnmatch
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Final, cast
+
+try:
+    import jsonschema
+except ImportError as exc:
+    raise SystemExit(
+        f"缺少依赖 jsonschema，请使用 uvx --with jsonschema --with pyyaml python <script>\n{exc}"
+    ) from exc
+
+try:
+    import yaml
+except ImportError as exc:
+    raise SystemExit(
+        f"缺少依赖 PyYAML，请使用 uvx --with pyyaml --with jsonschema python <script>\n{exc}"
+    ) from exc
+
+
+JSONMapping = dict[str, object]
+
+SOURCE_CONFIG_SCHEMA_RELATIVE: Final[Path] = Path(
+    "platform/contracts/knowledge-source-config.schema.yaml"
+)
+ARTIFACT_SCHEMA_RELATIVE: Final[Path] = Path(
+    "platform/contracts/knowledge-artifact.schema.yaml"
+)
+PUBLISH_RECORD_SCHEMA_RELATIVE: Final[Path] = Path(
+    "platform/contracts/knowledge-publish-record.schema.yaml"
+)
+DOCUMENT_SCHEMA_RELATIVE: Final[Path] = Path(
+    "platform/contracts/knowledge-document.schema.yaml"
+)
+CHUNK_SCHEMA_RELATIVE: Final[Path] = Path(
+    "platform/contracts/knowledge-chunk.schema.yaml"
+)
+SOURCE_REGISTRY_RELATIVE: Final[Path] = Path("platform/knowledge/source-registry.yaml")
+ACL_POLICIES_RELATIVE: Final[Path] = Path("platform/knowledge/acl-policies.yaml")
+PUBLISH_TARGETS_RELATIVE: Final[Path] = Path("platform/knowledge/publish-targets.yaml")
+DEFAULT_MAX_CHUNK_CHARACTERS: Final[int] = 1200
+
+
+class KnowledgeContractError(Exception):
+    """Raised when a knowledge contract is invalid."""
+
+
+@dataclass(frozen=True, slots=True)
+class ChangeEntry:
+    """Represents one git diff path change."""
+
+    change_type: str
+    path: str
+
+
+@dataclass(frozen=True, slots=True)
+class SourceRule:
+    """Represents one source rule declared by a business repository."""
+
+    source_id: str
+    include: tuple[str, ...]
+    exclude: tuple[str, ...]
+    visibility: str
+    source_type: str
+    doc_type: str
+    product_area: str
+    owner: str
+    review_status: str
+    tags: tuple[str, ...]
+    title_strategy: str
+    title_prefix: str
+    approved_by: str | None
+    effective_from: str | None
+    expires_at: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class KnowledgePaths:
+    """Filesystem layout inside ljwx-deploy for knowledge state."""
+
+    artifacts_root: Path
+    events_path: Path
+    publish_records_root: Path
+
+
+def utc_now() -> str:
+    """Return an ISO-8601 timestamp in UTC."""
+
+    return datetime.now(tz=UTC).isoformat()
+
+
+def ensure_directory(path: Path) -> None:
+    """Create a directory tree when it does not exist."""
+
+    path.mkdir(parents=True, exist_ok=True)
+
+
+def ensure_parent(path: Path) -> None:
+    """Create the parent directory for a file path."""
+
+    ensure_directory(path.parent)
+
+
+def write_json(path: Path, payload: JSONMapping | list[JSONMapping]) -> None:
+    """Write a JSON file with deterministic formatting."""
+
+    ensure_parent(path)
+    path.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+
+
+def load_json_mapping(path: Path) -> JSONMapping:
+    """Load a JSON object from disk."""
+
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(raw, dict):
+        raise KnowledgeContractError(f"{path} 根节点必须是 object")
+    return {str(key): value for key, value in raw.items()}
+
+
+def load_yaml_mapping(path: Path) -> JSONMapping:
+    """Load a YAML mapping from disk."""
+
+    raw = yaml.safe_load(path.read_text(encoding="utf-8"))
+    if raw is None:
+        return {}
+    if not isinstance(raw, dict):
+        raise KnowledgeContractError(f"{path} 根节点必须是 mapping")
+    return {str(key): value for key, value in raw.items()}
+
+
+def write_jsonl(path: Path, rows: list[JSONMapping]) -> None:
+    """Write JSONL rows to disk."""
+
+    ensure_parent(path)
+    with path.open("w", encoding="utf-8") as handle:
+        for row in rows:
+            handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True))
+            handle.write("\n")
+
+
+def append_jsonl(path: Path, row: JSONMapping) -> None:
+    """Append one JSONL row to disk."""
+
+    ensure_parent(path)
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True))
+        handle.write("\n")
+
+
+def load_jsonl(path: Path) -> list[JSONMapping]:
+    """Load JSONL records from disk when present."""
+
+    if not path.exists():
+        return []
+    rows: list[JSONMapping] = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        stripped = line.strip()
+        if not stripped:
+            continue
+        raw = json.loads(stripped)
+        if not isinstance(raw, dict):
+            raise KnowledgeContractError(f"{path} 存在非 mapping JSONL 记录")
+        rows.append({str(key): value for key, value in raw.items()})
+    return rows
+
+
+def load_validator(path: Path) -> jsonschema.Draft202012Validator:
+    """Load a JSON schema validator from YAML."""
+
+    schema = load_yaml_mapping(path)
+    jsonschema.Draft202012Validator.check_schema(schema)
+    return jsonschema.Draft202012Validator(schema)
+
+
+def validate_payload(
+    validator: jsonschema.Draft202012Validator,
+    payload: JSONMapping,
+    payload_name: str,
+) -> None:
+    """Validate one payload against a schema."""
+
+    try:
+        validator.validate(payload)
+    except jsonschema.ValidationError as exc:
+        raise KnowledgeContractError(
+            f"{payload_name} 不符合 schema: {exc.message}"
+        ) from exc
+
+
+def require_mapping(value: object, field_path: str) -> JSONMapping:
+    """Require a JSON mapping."""
+
+    if not isinstance(value, dict):
+        raise KnowledgeContractError(f"{field_path} 必须是 mapping")
+    return {str(key): item for key, item in value.items()}
+
+
+def require_list(value: object, field_path: str) -> list[object]:
+    """Require a list."""
+
+    if not isinstance(value, list):
+        raise KnowledgeContractError(f"{field_path} 必须是列表")
+    return value
+
+
+def require_non_empty_str(value: object, field_path: str) -> str:
+    """Require a non-empty string."""
+
+    if not isinstance(value, str) or not value.strip():
+        raise KnowledgeContractError(f"{field_path} 必须是非空字符串")
+    return value.strip()
+
+
+def optional_non_empty_str(value: object, field_path: str) -> str | None:
+    """Read an optional non-empty string."""
+
+    if value in ("", None):
+        return None
+    return require_non_empty_str(value, field_path)
+
+
+def require_str_list(value: object, field_path: str) -> list[str]:
+    """Require a list of strings."""
+
+    items = require_list(value, field_path)
+    results: list[str] = []
+    for index, item in enumerate(items):
+        results.append(require_non_empty_str(item, f"{field_path}[{index}]"))
+    return results
+
+
+def sha256_text(text: str) -> str:
+    """Return the SHA-256 digest of a text payload."""
+
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def slugify_fragment(value: str) -> str:
+    """Convert a string into a filesystem-safe slug."""
+
+    normalized = re.sub(r"[^a-zA-Z0-9]+", "-", value).strip("-").lower()
+    return normalized or "unknown"
+
+
+def repo_slug(repo: str) -> str:
+    """Return a stable directory slug for an owner/repo name."""
+
+    return repo.replace("/", "__")
+
+
+def build_document_id(source_repo: str, source_path: str) -> str:
+    """Build a deterministic document identifier from repo and path."""
+
+    repo_name = source_repo.split("/")[-1]
+    base_slug = slugify_fragment(f"{repo_name}-{source_path}")
+    if len(base_slug) <= 96:
+        return base_slug
+    digest = hashlib.sha1(f"{source_repo}:{source_path}".encode("utf-8")).hexdigest()[
+        :12
+    ]
+    return f"{base_slug[:80]}-{digest}"
+
+
+def match_globs(
+    path: str, includes: tuple[str, ...], excludes: tuple[str, ...]
+) -> bool:
+    """Return True when a path matches include/exclude rules."""
+
+    included = any(fnmatch.fnmatch(path, pattern) for pattern in includes)
+    if not included:
+        return False
+    return not any(fnmatch.fnmatch(path, pattern) for pattern in excludes)
+
+
+def parse_diff_entries(path: Path) -> list[ChangeEntry]:
+    """Parse git diff --name-status output into flattened path changes."""
+
+    if not path.exists():
+        return []
+
+    results: list[ChangeEntry] = []
+    for raw_line in path.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        parts = line.split("\t")
+        status = require_non_empty_str(parts[0], "diff.status")
+        base_status = status[0]
+        if base_status in {"A", "C", "M", "T"}:
+            if len(parts) < 2:
+                raise KnowledgeContractError(f"无效 diff 行: {line}")
+            results.append(ChangeEntry(change_type="upsert", path=parts[-1]))
+            continue
+        if base_status == "D":
+            if len(parts) != 2:
+                raise KnowledgeContractError(f"无效删除 diff 行: {line}")
+            results.append(ChangeEntry(change_type="delete", path=parts[1]))
+            continue
+        if base_status == "R":
+            if len(parts) != 3:
+                raise KnowledgeContractError(f"无效重命名 diff 行: {line}")
+            results.append(ChangeEntry(change_type="delete", path=parts[1]))
+            results.append(ChangeEntry(change_type="upsert", path=parts[2]))
+            continue
+        raise KnowledgeContractError(f"暂不支持的 diff 状态: {status}")
+    return results
+
+
+def extract_markdown_title(content: str, source_path: str, strategy: str) -> str:
+    """Resolve a document title from markdown content."""
+
+    if strategy == "filename":
+        return Path(source_path).stem.replace("-", " ").replace("_", " ").strip()
+
+    for line in content.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("# "):
+            return stripped.removeprefix("# ").strip()
+
+    return Path(source_path).stem.replace("-", " ").replace("_", " ").strip()
+
+
+def chunk_markdown(
+    content: str, max_characters: int = DEFAULT_MAX_CHUNK_CHARACTERS
+) -> list[str]:
+    """Chunk markdown into paragraph groups."""
+
+    paragraphs = [item.strip() for item in content.split("\n\n") if item.strip()]
+    if not paragraphs:
+        return []
+
+    chunks: list[str] = []
+    current_parts: list[str] = []
+    current_length = 0
+    for paragraph in paragraphs:
+        paragraph_length = len(paragraph)
+        if current_parts and current_length + paragraph_length > max_characters:
+            chunks.append("\n\n".join(current_parts))
+            current_parts = [paragraph]
+            current_length = paragraph_length
+            continue
+        current_parts.append(paragraph)
+        current_length += paragraph_length
+
+    if current_parts:
+        chunks.append("\n\n".join(current_parts))
+
+    return chunks
+
+
+def knowledge_paths(deploy_root: Path, environment: str) -> KnowledgePaths:
+    """Resolve the filesystem layout used by central knowledge state."""
+
+    base_root = deploy_root / "knowledge"
+    return KnowledgePaths(
+        artifacts_root=base_root / "artifacts" / environment,
+        events_path=base_root / "events" / environment / "events.jsonl",
+        publish_records_root=base_root / "publish-records" / environment,
+    )
+
+
+def load_source_rules(
+    config_path: Path, deploy_root: Path
+) -> tuple[str, list[SourceRule]]:
+    """Load and validate source config rules."""
+
+    validator = load_validator(deploy_root / SOURCE_CONFIG_SCHEMA_RELATIVE)
+    payload = load_yaml_mapping(config_path)
+    validate_payload(validator, payload, config_path.as_posix())
+
+    service = require_non_empty_str(
+        payload.get("service"), "knowledge.sources.yaml.service"
+    )
+    raw_sources = require_list(payload.get("sources"), "knowledge.sources.yaml.sources")
+
+    rules: list[SourceRule] = []
+    for index, raw_source in enumerate(raw_sources):
+        source = require_mapping(raw_source, f"knowledge.sources.yaml.sources[{index}]")
+        rules.append(
+            SourceRule(
+                source_id=require_non_empty_str(
+                    source.get("id"), f"knowledge.sources.yaml.sources[{index}].id"
+                ),
+                include=tuple(
+                    require_str_list(
+                        source.get("include"),
+                        f"knowledge.sources.yaml.sources[{index}].include",
+                    )
+                ),
+                exclude=tuple(
+                    require_str_list(
+                        source.get("exclude", []),
+                        f"knowledge.sources.yaml.sources[{index}].exclude",
+                    )
+                ),
+                visibility=require_non_empty_str(
+                    source.get("visibility"),
+                    f"knowledge.sources.yaml.sources[{index}].visibility",
+                ),
+                source_type=require_non_empty_str(
+                    source.get("source_type"),
+                    f"knowledge.sources.yaml.sources[{index}].source_type",
+                ),
+                doc_type=require_non_empty_str(
+                    source.get("doc_type"),
+                    f"knowledge.sources.yaml.sources[{index}].doc_type",
+                ),
+                product_area=require_non_empty_str(
+                    source.get("product_area"),
+                    f"knowledge.sources.yaml.sources[{index}].product_area",
+                ),
+                owner=require_non_empty_str(
+                    source.get("owner"),
+                    f"knowledge.sources.yaml.sources[{index}].owner",
+                ),
+                review_status=require_non_empty_str(
+                    source.get("review_status"),
+                    f"knowledge.sources.yaml.sources[{index}].review_status",
+                ),
+                tags=tuple(
+                    require_str_list(
+                        source.get("tags", []),
+                        f"knowledge.sources.yaml.sources[{index}].tags",
+                    )
+                ),
+                title_strategy=optional_non_empty_str(
+                    source.get("title_strategy"),
+                    f"knowledge.sources.yaml.sources[{index}].title_strategy",
+                )
+                or "first_heading",
+                title_prefix=optional_non_empty_str(
+                    source.get("title_prefix"),
+                    f"knowledge.sources.yaml.sources[{index}].title_prefix",
+                )
+                or "",
+                approved_by=optional_non_empty_str(
+                    source.get("approved_by"),
+                    f"knowledge.sources.yaml.sources[{index}].approved_by",
+                ),
+                effective_from=optional_non_empty_str(
+                    source.get("effective_from"),
+                    f"knowledge.sources.yaml.sources[{index}].effective_from",
+                ),
+                expires_at=optional_non_empty_str(
+                    source.get("expires_at"),
+                    f"knowledge.sources.yaml.sources[{index}].expires_at",
+                ),
+            )
+        )
+    return service, rules
+
+
+def load_source_registry(deploy_root: Path) -> dict[str, dict[str, object]]:
+    """Load artifact source registry entries keyed by repository."""
+
+    payload = load_yaml_mapping(deploy_root / SOURCE_REGISTRY_RELATIVE)
+    raw_registry = require_list(
+        payload.get("artifact_sources", []),
+        "platform/knowledge/source-registry.yaml.artifact_sources",
+    )
+    registry: dict[str, dict[str, object]] = {}
+    for index, raw_entry in enumerate(raw_registry):
+        entry = require_mapping(
+            raw_entry,
+            f"platform/knowledge/source-registry.yaml.artifact_sources[{index}]",
+        )
+        repo_name = require_non_empty_str(
+            entry.get("repo"),
+            f"platform/knowledge/source-registry.yaml.artifact_sources[{index}].repo",
+        )
+        registry[repo_name] = {
+            "service": require_non_empty_str(
+                entry.get("service"),
+                "platform/knowledge/source-registry.yaml.artifact_sources."
+                f"[{index}].service",
+            ),
+            "source_ids": require_str_list(
+                entry.get("source_ids", []),
+                "platform/knowledge/source-registry.yaml.artifact_sources."
+                f"[{index}].source_ids",
+            ),
+            "allowed_visibilities": require_str_list(
+                entry.get("allowed_visibilities", []),
+                "platform/knowledge/source-registry.yaml.artifact_sources."
+                f"[{index}].allowed_visibilities",
+            ),
+        }
+    return registry
+
+
+def load_acl_policies(deploy_root: Path) -> dict[str, dict[str, object]]:
+    """Load ACL policies keyed by visibility."""
+
+    payload = load_yaml_mapping(deploy_root / ACL_POLICIES_RELATIVE)
+    raw_policies = require_mapping(
+        payload.get("policies"),
+        "platform/knowledge/acl-policies.yaml.policies",
+    )
+    policies: dict[str, dict[str, object]] = {}
+    for visibility, raw_policy in raw_policies.items():
+        policy = require_mapping(
+            raw_policy,
+            f"platform/knowledge/acl-policies.yaml.policies.{visibility}",
+        )
+        policies[visibility] = {
+            "acl_expression_template": require_non_empty_str(
+                policy.get("acl_expression_template"),
+                "platform/knowledge/acl-policies.yaml.policies."
+                f"{visibility}.acl_expression_template",
+            ),
+            "allowed_targets": require_str_list(
+                policy.get("allowed_targets", []),
+                "platform/knowledge/acl-policies.yaml.policies."
+                f"{visibility}.allowed_targets",
+            ),
+            "review_required": bool(policy.get("review_required", False)),
+        }
+    return policies
+
+
+def load_publish_targets(deploy_root: Path) -> dict[str, dict[str, object]]:
+    """Load publish targets keyed by target id."""
+
+    payload = load_yaml_mapping(deploy_root / PUBLISH_TARGETS_RELATIVE)
+    raw_targets = require_mapping(
+        payload.get("targets"),
+        "platform/knowledge/publish-targets.yaml.targets",
+    )
+    targets: dict[str, dict[str, object]] = {}
+    for target_id, raw_target in raw_targets.items():
+        target = require_mapping(
+            raw_target, f"platform/knowledge/publish-targets.yaml.targets.{target_id}"
+        )
+        targets[target_id] = {
+            "type": require_non_empty_str(
+                target.get("type"),
+                f"platform/knowledge/publish-targets.yaml.targets.{target_id}.type",
+            ),
+            "visibility": require_non_empty_str(
+                target.get("visibility"),
+                "platform/knowledge/publish-targets.yaml.targets."
+                f"{target_id}.visibility",
+            ),
+            "dataset_name": optional_non_empty_str(
+                target.get("dataset_name"),
+                "platform/knowledge/publish-targets.yaml.targets."
+                f"{target_id}.dataset_name",
+            ),
+        }
+    return targets
+
+
+def load_publish_record(path: Path) -> JSONMapping:
+    """Load one publish record file."""
+
+    return load_json_mapping(path)
+
+
+def load_publish_records(root: Path) -> dict[str, JSONMapping]:
+    """Load all publish records under a directory."""
+
+    if not root.exists():
+        return {}
+    records: dict[str, JSONMapping] = {}
+    for path in sorted(root.rglob("*.json")):
+        record = load_publish_record(path)
+        document_id = require_non_empty_str(
+            record.get("document_id"), f"{path}.document_id"
+        )
+        records[document_id] = record
+    return records
+
+
+def build_acl_expression(
+    visibility: str, policies: dict[str, dict[str, object]]
+) -> str:
+    """Resolve the ACL expression for a visibility."""
+
+    policy = policies.get(visibility)
+    if policy is None:
+        raise KnowledgeContractError(f"未注册 visibility ACL policy: {visibility}")
+    acl_template = policy.get("acl_expression_template")
+    return require_non_empty_str(acl_template, f"acl policy for {visibility}")
+
+
+def build_event(
+    *,
+    action: str,
+    artifact_id: str,
+    document_id: str,
+    source_repo: str,
+    status: str,
+    reason: str,
+    target_ids: list[str],
+) -> JSONMapping:
+    """Build one append-only event record."""
+
+    compact_time = utc_now().replace(":", "").replace("-", "")
+    return {
+        "event_id": f"knowledge-{action}-{document_id}-{compact_time}",
+        "timestamp": utc_now(),
+        "artifact_id": artifact_id,
+        "document_id": document_id,
+        "source_repo": source_repo,
+        "action": action,
+        "status": status,
+        "reason": reason,
+        "target_ids": target_ids,
+    }
+
+
+def artifact_document_name(document_id: str) -> str:
+    """Return the stable remote document name used in Dify."""
+
+    return document_id
+
+
+def artifact_snapshot_root(
+    deploy_root: Path,
+    environment: str,
+    source_repo: str,
+    artifact_id: str,
+) -> Path:
+    """Return the snapshot root for one artifact."""
+
+    return (
+        knowledge_paths(deploy_root, environment).artifacts_root
+        / repo_slug(source_repo)
+        / artifact_id
+    )
+
+
+def record_path_for_document(
+    deploy_root: Path,
+    environment: str,
+    source_repo: str,
+    document_id: str,
+) -> Path:
+    """Return the publish-record path for a document."""
+
+    return (
+        knowledge_paths(deploy_root, environment).publish_records_root
+        / repo_slug(source_repo)
+        / f"{document_id}.json"
+    )
+
+
+def cast_mapping_list(value: object, field_path: str) -> list[JSONMapping]:
+    """Cast a list payload into a list of mappings."""
+
+    items = require_list(value, field_path)
+    mappings: list[JSONMapping] = []
+    for index, item in enumerate(items):
+        mappings.append(require_mapping(item, f"{field_path}[{index}]"))
+    return mappings
+
+
+def optional_mapping(value: object, field_path: str) -> JSONMapping | None:
+    """Return an optional mapping."""
+
+    if value is None:
+        return None
+    return require_mapping(value, field_path)
+
+
+def cast_str(value: object) -> str:
+    """Cast a value known to be a string."""
+
+    return cast(str, value)

scripts/knowledge/dify_client.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+"""Minimal Dify service API client for knowledge publication."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Final
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+from .common import (
+    JSONMapping,
+    optional_non_empty_str,
+    require_mapping,
+    require_non_empty_str,
+)
+
+
+REQUEST_TIMEOUT_SECONDS: Final[int] = 15
+
+
+class DifyClientError(Exception):
+    """Raised when a Dify API request fails."""
+
+
+@dataclass(frozen=True, slots=True)
+class DifyDataset:
+    """Dify dataset summary."""
+
+    dataset_id: str
+    name: str
+
+
+@dataclass(frozen=True, slots=True)
+class DifyDocument:
+    """Dify document summary."""
+
+    document_id: str
+    name: str
+
+
+class DifyClient:
+    """Minimal service API client for Dify dataset operations."""
+
+    def __init__(self, base_url: str, api_key: str) -> None:
+        if not base_url.strip():
+            raise DifyClientError("缺少 Dify base URL")
+        if not api_key.strip():
+            raise DifyClientError("缺少 Dify API key")
+        self._base_url = base_url.rstrip("/")
+        self._api_key = api_key
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        payload: JSONMapping | None = None,
+        query: dict[str, str] | None = None,
+        expect_empty: bool = False,
+    ) -> JSONMapping:
+        url = f"{self._base_url}{path}"
+        if query:
+            url = f"{url}?{urlencode(query)}"
+
+        data: bytes | None = None
+        headers = {
+            "Authorization": f"Bearer {self._api_key}",
+            "Content-Type": "application/json",
+        }
+        if payload is not None:
+            data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+
+        request = Request(url=url, data=data, method=method, headers=headers)
+        try:
+            with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+                raw_body = response.read().decode("utf-8").strip()
+        except HTTPError as exc:
+            error_body = exc.read().decode("utf-8", errors="replace")
+            raise DifyClientError(
+                f"Dify 请求失败: {method} {path} -> HTTP {exc.code}: {error_body}"
+            ) from exc
+        except URLError as exc:
+            raise DifyClientError(
+                f"Dify 请求失败: {method} {path} -> {exc.reason}"
+            ) from exc
+
+        if expect_empty and not raw_body:
+            return {}
+        if not raw_body:
+            return {}
+        try:
+            payload_object = json.loads(raw_body)
+        except json.JSONDecodeError as exc:
+            raise DifyClientError(f"Dify 返回了无效 JSON: {raw_body}") from exc
+        if not isinstance(payload_object, dict):
+            raise DifyClientError("Dify 返回根节点必须是 object")
+        return {str(key): value for key, value in payload_object.items()}
+
+    def list_datasets(self) -> list[DifyDataset]:
+        """List Dify datasets visible to the service API token."""
+
+        payload = self._request("GET", "/datasets", query={"page": "1", "limit": "100"})
+        raw_items = payload.get("data", [])
+        if not isinstance(raw_items, list):
+            raise DifyClientError("/datasets.data 必须是列表")
+
+        results: list[DifyDataset] = []
+        for index, raw_item in enumerate(raw_items):
+            item = require_mapping(raw_item, f"/datasets.data[{index}]")
+            results.append(
+                DifyDataset(
+                    dataset_id=require_non_empty_str(
+                        item.get("id"), f"/datasets.data[{index}].id"
+                    ),
+                    name=require_non_empty_str(
+                        item.get("name"), f"/datasets.data[{index}].name"
+                    ),
+                )
+            )
+        return results
+
+    def ensure_dataset(self, dataset_name: str) -> DifyDataset:
+        """Return an existing dataset or create it on demand."""
+
+        for dataset in self.list_datasets():
+            if dataset.name == dataset_name:
+                return dataset
+
+        payload = self._request(
+            "POST",
+            "/datasets",
+            payload={
+                "name": dataset_name,
+                "description": f"中央知识流水线管理的 {dataset_name} dataset",
+                "indexing_technique": "high_quality",
+                "permission": "only_me",
+            },
+        )
+        dataset_id = require_non_empty_str(payload.get("id"), "POST /datasets.id")
+        created_name = require_non_empty_str(payload.get("name"), "POST /datasets.name")
+        return DifyDataset(dataset_id=dataset_id, name=created_name)
+
+    def list_documents(self, dataset_id: str) -> list[DifyDocument]:
+        """List documents in one dataset."""
+
+        payload = self._request(
+            "GET",
+            f"/datasets/{dataset_id}/documents",
+            query={"page": "1", "limit": "100"},
+        )
+        raw_items = payload.get("data", [])
+        if not isinstance(raw_items, list):
+            raise DifyClientError("/documents.data 必须是列表")
+
+        results: list[DifyDocument] = []
+        for index, raw_item in enumerate(raw_items):
+            item = require_mapping(
+                raw_item, f"/datasets/{dataset_id}/documents.data[{index}]"
+            )
+            results.append(
+                DifyDocument(
+                    document_id=require_non_empty_str(
+                        item.get("id"),
+                        f"/datasets/{dataset_id}/documents.data[{index}].id",
+                    ),
+                    name=require_non_empty_str(
+                        item.get("name"),
+                        f"/datasets/{dataset_id}/documents.data[{index}].name",
+                    ),
+                )
+            )
+        return results
+
+    def find_document_id(self, dataset_id: str, document_name: str) -> str | None:
+        """Find a document by its stable name."""
+
+        for document in self.list_documents(dataset_id):
+            if document.name == document_name:
+                return document.document_id
+        return None
+
+    def upsert_document(
+        self,
+        *,
+        dataset_id: str,
+        document_name: str,
+        content: str,
+        document_id: str | None,
+    ) -> str:
+        """Create or update one Dify text document."""
+
+        if document_id:
+            payload = self._request(
+                "POST",
+                f"/datasets/{dataset_id}/documents/{document_id}/update_by_text",
+                payload={
+                    "name": document_name,
+                    "text": content,
+                    "process_rule": {"mode": "automatic"},
+                },
+            )
+            updated_id = optional_non_empty_str(
+                payload.get("document_id"),
+                f"/datasets/{dataset_id}/documents/{document_id}/update_by_text.document_id",
+            )
+            return updated_id or document_id
+
+        payload = self._request(
+            "POST",
+            f"/datasets/{dataset_id}/document/create_by_text",
+            payload={
+                "name": document_name,
+                "text": content,
+                "indexing_technique": "high_quality",
+                "process_rule": {"mode": "automatic"},
+            },
+        )
+        raw_document = require_mapping(
+            payload.get("document"),
+            f"/datasets/{dataset_id}/document/create_by_text.document",
+        )
+        return require_non_empty_str(
+            raw_document.get("id"),
+            f"/datasets/{dataset_id}/document/create_by_text.document.id",
+        )
+
+    def delete_document(self, dataset_id: str, document_id: str) -> None:
+        """Delete one Dify document."""
+
+        self._request(
+            "DELETE",
+            f"/datasets/{dataset_id}/documents/{document_id}",
+            expect_empty=True,
+        )

scripts/knowledge/export_artifact.py

@@ -0,0 +1,343 @@
+#!/usr/bin/env python3
+"""Export markdown knowledge sources into a portable artifact."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import tempfile
+import zipfile
+from pathlib import Path
+
+from .common import (
+    ARTIFACT_SCHEMA_RELATIVE,
+    JSONMapping,
+    KnowledgeContractError,
+    build_document_id,
+    extract_markdown_title,
+    load_source_rules,
+    load_validator,
+    match_globs,
+    parse_diff_entries,
+    repo_slug,
+    require_non_empty_str,
+    sha256_text,
+    utc_now,
+    validate_payload,
+    write_json,
+)
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse CLI arguments."""
+
+    parser = argparse.ArgumentParser(description="Export a knowledge artifact")
+    parser.add_argument(
+        "--repo-root", type=Path, required=True, help="Business repo root"
+    )
+    parser.add_argument(
+        "--deploy-root",
+        type=Path,
+        required=True,
+        help="ljwx-deploy root used for contracts",
+    )
+    parser.add_argument(
+        "--config",
+        type=Path,
+        default=Path("knowledge.sources.yaml"),
+        help="Path to knowledge.sources.yaml relative to repo root",
+    )
+    parser.add_argument("--source-repo", required=True, help="GitHub owner/repo")
+    parser.add_argument("--source-ref", required=True, help="Git ref")
+    parser.add_argument("--source-sha", required=True, help="Git commit SHA")
+    parser.add_argument("--workflow-name", required=True, help="Workflow name")
+    parser.add_argument("--run-id", type=int, required=True, help="GitHub run id")
+    parser.add_argument(
+        "--run-attempt", type=int, required=True, help="GitHub run attempt"
+    )
+    parser.add_argument(
+        "--changed-files",
+        type=Path,
+        help="Path to git diff --name-status output. If omitted, export all matches.",
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        required=True,
+        help="Artifact zip output path",
+    )
+    parser.add_argument(
+        "--summary-json",
+        type=Path,
+        required=True,
+        help="Where to write export summary JSON",
+    )
+    return parser.parse_args()
+
+
+def resolve_rule_for_path(path: str, rules: list[object]) -> object | None:
+    """Find a single matching rule for a source path."""
+
+    matched: list[object] = []
+    for rule in rules:
+        if not hasattr(rule, "include") or not hasattr(rule, "exclude"):
+            continue
+        include = getattr(rule, "include")
+        exclude = getattr(rule, "exclude")
+        if not isinstance(include, tuple) or not isinstance(exclude, tuple):
+            continue
+        if match_globs(path, include, exclude):
+            matched.append(rule)
+    if len(matched) > 1:
+        raise KnowledgeContractError(f"路径命中了多个 source rule: {path}")
+    return matched[0] if matched else None
+
+
+def build_manifest_document(
+    *,
+    repo_root: Path,
+    source_repo: str,
+    source_sha: str,
+    relative_path: str,
+    source_id: str,
+    visibility: str,
+    source_type: str,
+    doc_type: str,
+    product_area: str,
+    owner: str,
+    review_status: str,
+    title_strategy: str,
+    title_prefix: str,
+    tags: tuple[str, ...],
+    approved_by: str | None,
+    effective_from: str | None,
+    expires_at: str | None,
+) -> tuple[JSONMapping, str, str]:
+    """Build one upsert artifact document entry and its content payload."""
+
+    file_path = repo_root / relative_path
+    content = file_path.read_text(encoding="utf-8")
+    document_id = build_document_id(source_repo, relative_path)
+    title = extract_markdown_title(content, relative_path, title_strategy)
+    if title_prefix:
+        title = f"{title_prefix}{title}"
+    checksum = sha256_text(content)
+    content_path = f"documents/{document_id}.md"
+    source_ref = f"{source_repo}@{source_sha}:{relative_path}"
+    payload: JSONMapping = {
+        "op": "upsert",
+        "document_id": document_id,
+        "source_id": source_id,
+        "source_path": relative_path,
+        "source_ref": source_ref,
+        "content_path": content_path,
+        "title": title,
+        "visibility": visibility,
+        "source_type": source_type,
+        "doc_type": doc_type,
+        "product_area": product_area,
+        "owner": owner,
+        "review_status": review_status,
+        "version": source_sha,
+        "checksum": checksum,
+        "tags": list(tags),
+        "approved_by": approved_by,
+        "effective_from": effective_from,
+        "expires_at": expires_at,
+        "reason": None,
+    }
+    return payload, content_path, content
+
+
+def build_invalidate_document(
+    *,
+    source_repo: str,
+    source_sha: str,
+    relative_path: str,
+    source_id: str,
+) -> JSONMapping:
+    """Build one invalidate artifact document entry."""
+
+    return {
+        "op": "invalidate",
+        "document_id": build_document_id(source_repo, relative_path),
+        "source_id": source_id,
+        "source_path": relative_path,
+        "source_ref": f"{source_repo}@{source_sha}:{relative_path}",
+        "content_path": None,
+        "title": None,
+        "visibility": None,
+        "source_type": None,
+        "doc_type": None,
+        "product_area": None,
+        "owner": None,
+        "review_status": None,
+        "version": None,
+        "checksum": None,
+        "tags": [],
+        "approved_by": None,
+        "effective_from": None,
+        "expires_at": None,
+        "reason": "source_deleted",
+    }
+
+
+def main() -> int:
+    """CLI entry point."""
+
+    args = parse_args()
+    repo_root = args.repo_root.resolve()
+    deploy_root = args.deploy_root.resolve()
+    config_path = (repo_root / args.config).resolve()
+    output_path = args.output.resolve()
+    summary_path = args.summary_json.resolve()
+
+    service, rules = load_source_rules(config_path, deploy_root)
+    artifact_validator = load_validator(deploy_root / ARTIFACT_SCHEMA_RELATIVE)
+    diff_entries = (
+        parse_diff_entries(args.changed_files.resolve()) if args.changed_files else []
+    )
+
+    changed_upserts = {
+        entry.path for entry in diff_entries if entry.change_type == "upsert"
+    }
+    changed_deletes = {
+        entry.path for entry in diff_entries if entry.change_type == "delete"
+    }
+    export_all = not diff_entries
+
+    live_documents: list[JSONMapping] = []
+    deleted_documents: list[JSONMapping] = []
+    staged_documents: list[tuple[str, str]] = []
+
+    if export_all:
+        candidate_paths = sorted(
+            path.relative_to(repo_root).as_posix()
+            for path in repo_root.rglob("*.md")
+            if path.is_file()
+        )
+    else:
+        candidate_paths = sorted(changed_upserts)
+
+    for relative_path in candidate_paths:
+        rule = resolve_rule_for_path(relative_path, rules)
+        if rule is None:
+            continue
+        rule_source_id = require_non_empty_str(
+            getattr(rule, "source_id"), "rule.source_id"
+        )
+        payload, content_path, content = build_manifest_document(
+            repo_root=repo_root,
+            source_repo=args.source_repo,
+            source_sha=args.source_sha,
+            relative_path=relative_path,
+            source_id=rule_source_id,
+            visibility=require_non_empty_str(
+                getattr(rule, "visibility"), "rule.visibility"
+            ),
+            source_type=require_non_empty_str(
+                getattr(rule, "source_type"), "rule.source_type"
+            ),
+            doc_type=require_non_empty_str(getattr(rule, "doc_type"), "rule.doc_type"),
+            product_area=require_non_empty_str(
+                getattr(rule, "product_area"), "rule.product_area"
+            ),
+            owner=require_non_empty_str(getattr(rule, "owner"), "rule.owner"),
+            review_status=require_non_empty_str(
+                getattr(rule, "review_status"), "rule.review_status"
+            ),
+            title_strategy=require_non_empty_str(
+                getattr(rule, "title_strategy"), "rule.title_strategy"
+            ),
+            title_prefix=require_non_empty_str(
+                getattr(rule, "title_prefix", ""), "rule.title_prefix"
+            )
+            if getattr(rule, "title_prefix", "") != ""
+            else "",
+            tags=getattr(rule, "tags"),
+            approved_by=getattr(rule, "approved_by"),
+            effective_from=getattr(rule, "effective_from"),
+            expires_at=getattr(rule, "expires_at"),
+        )
+        live_documents.append(payload)
+        staged_documents.append((content_path, content))
+
+    for relative_path in sorted(changed_deletes):
+        rule = resolve_rule_for_path(relative_path, rules)
+        if rule is None:
+            continue
+        deleted_documents.append(
+            build_invalidate_document(
+                source_repo=args.source_repo,
+                source_sha=args.source_sha,
+                relative_path=relative_path,
+                source_id=require_non_empty_str(
+                    getattr(rule, "source_id"), "rule.source_id"
+                ),
+            )
+        )
+
+    all_documents = deleted_documents + live_documents
+    artifact_id = f"{repo_slug(args.source_repo)}-{args.source_sha[:12]}"
+    summary: JSONMapping = {
+        "artifact_id": artifact_id,
+        "artifact_file": str(output_path),
+        "deleted_documents": len(deleted_documents),
+        "exported": bool(all_documents),
+        "service": service,
+        "source_repo": args.source_repo,
+        "upsert_documents": len(live_documents),
+    }
+
+    if not all_documents:
+        write_json(summary_path, summary)
+        return 0
+
+    manifest: JSONMapping = {
+        "schema_version": "0.1.0",
+        "artifact_id": artifact_id,
+        "generated_at": utc_now(),
+        "export_mode": "full" if export_all else "changed",
+        "source": {
+            "repo": args.source_repo,
+            "service": service,
+            "commit": args.source_sha,
+            "ref": args.source_ref,
+            "workflow": args.workflow_name,
+            "run_id": args.run_id,
+            "run_attempt": args.run_attempt,
+        },
+        "documents": all_documents,
+    }
+    validate_payload(artifact_validator, manifest, "knowledge artifact manifest")
+
+    staging_root = Path(
+        tempfile.mkdtemp(prefix="knowledge-export-", dir=output_path.parent.as_posix())
+    )
+    try:
+        manifest_path = staging_root / "manifest.json"
+        write_json(manifest_path, manifest)
+        for content_path, content in staged_documents:
+            target_path = staging_root / content_path
+            target_path.parent.mkdir(parents=True, exist_ok=True)
+            target_path.write_text(content, encoding="utf-8")
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        with zipfile.ZipFile(
+            output_path, "w", compression=zipfile.ZIP_DEFLATED
+        ) as archive:
+            for path in sorted(staging_root.rglob("*")):
+                if path.is_dir():
+                    continue
+                archive.write(path, path.relative_to(staging_root))
+    finally:
+        shutil.rmtree(staging_root, ignore_errors=True)
+
+    write_json(summary_path, summary)
+    print(json.dumps(summary, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/knowledge/github_artifacts.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""GitHub Actions artifact helpers for the knowledge supply chain."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+from .common import JSONMapping, ensure_parent, require_mapping, require_non_empty_str
+
+
+REQUEST_TIMEOUT_SECONDS = 20
+
+
+class GitHubArtifactError(Exception):
+    """Raised when a GitHub artifact request fails."""
+
+
+def _request_json(url: str, token: str) -> JSONMapping:
+    request = Request(
+        url=url,
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": "2022-11-28",
+        },
+        method="GET",
+    )
+    try:
+        with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+            body = response.read().decode("utf-8")
+    except HTTPError as exc:
+        detail = exc.read().decode("utf-8", errors="replace")
+        raise GitHubArtifactError(
+            f"GitHub API 请求失败: HTTP {exc.code}: {detail}"
+        ) from exc
+    except URLError as exc:
+        raise GitHubArtifactError(f"GitHub API 请求失败: {exc.reason}") from exc
+
+    try:
+        payload = json.loads(body)
+    except json.JSONDecodeError as exc:
+        raise GitHubArtifactError(f"GitHub API 返回无效 JSON: {body}") from exc
+    if not isinstance(payload, dict):
+        raise GitHubArtifactError("GitHub API 返回根节点必须是 object")
+    return {str(key): value for key, value in payload.items()}
+
+
+def download_artifact_zip(
+    *,
+    repository: str,
+    run_id: int,
+    artifact_name: str,
+    token: str,
+    output_path: Path,
+) -> Path:
+    """Download one named artifact from a workflow run."""
+
+    artifacts_url = (
+        f"https://api.github.com/repos/{repository}/actions/runs/{run_id}/artifacts"
+    )
+    listing = _request_json(artifacts_url, token)
+    raw_items = listing.get("artifacts", [])
+    if not isinstance(raw_items, list):
+        raise GitHubArtifactError("/artifacts 必须是列表")
+
+    artifact_id: int | None = None
+    for index, raw_item in enumerate(raw_items):
+        item = require_mapping(raw_item, f"artifacts[{index}]")
+        candidate_name = require_non_empty_str(
+            item.get("name"), f"artifacts[{index}].name"
+        )
+        if candidate_name != artifact_name:
+            continue
+        raw_id = item.get("id")
+        if not isinstance(raw_id, int) or raw_id <= 0:
+            raise GitHubArtifactError(f"artifacts[{index}].id 必须是正整数")
+        artifact_id = raw_id
+        break
+
+    if artifact_id is None:
+        raise GitHubArtifactError(
+            f"未找到 artifact: repo={repository} run_id={run_id} name={artifact_name}"
+        )
+
+    download_url = (
+        f"https://api.github.com/repos/{repository}/actions/artifacts/{artifact_id}/zip"
+    )
+    request = Request(
+        url=download_url,
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": "2022-11-28",
+        },
+        method="GET",
+    )
+    try:
+        with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+            payload = response.read()
+    except HTTPError as exc:
+        detail = exc.read().decode("utf-8", errors="replace")
+        raise GitHubArtifactError(
+            f"下载 artifact 失败: HTTP {exc.code}: {detail}"
+        ) from exc
+    except URLError as exc:
+        raise GitHubArtifactError(f"下载 artifact 失败: {exc.reason}") from exc
+
+    ensure_parent(output_path)
+    output_path.write_bytes(payload)
+    return output_path

scripts/knowledge/ingest_artifact.py

@@ -0,0 +1,858 @@
+#!/usr/bin/env python3
+"""Consume one business-repo knowledge artifact in ljwx-deploy."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import tempfile
+import zipfile
+from pathlib import Path
+from typing import cast
+
+from .common import (
+    ARTIFACT_SCHEMA_RELATIVE,
+    CHUNK_SCHEMA_RELATIVE,
+    DOCUMENT_SCHEMA_RELATIVE,
+    JSONMapping,
+    KnowledgeContractError,
+    PUBLISH_RECORD_SCHEMA_RELATIVE,
+    append_jsonl,
+    artifact_document_name,
+    artifact_snapshot_root,
+    build_acl_expression,
+    build_event,
+    cast_mapping_list,
+    chunk_markdown,
+    knowledge_paths,
+    load_acl_policies,
+    load_json_mapping,
+    load_publish_records,
+    load_publish_targets,
+    load_source_registry,
+    load_validator,
+    optional_non_empty_str,
+    record_path_for_document,
+    require_mapping,
+    require_non_empty_str,
+    sha256_text,
+    utc_now,
+    validate_payload,
+    write_json,
+    write_jsonl,
+)
+from .dify_client import DifyClient, DifyClientError
+from .github_artifacts import GitHubArtifactError, download_artifact_zip
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse CLI arguments."""
+
+    parser = argparse.ArgumentParser(description="Ingest a knowledge artifact")
+    parser.add_argument(
+        "--deploy-root", type=Path, required=True, help="ljwx-deploy root"
+    )
+    parser.add_argument("--env", default="dev", help="Target environment")
+    parser.add_argument("--artifact-file", type=Path, help="Local artifact zip")
+    parser.add_argument("--source-repo", help="GitHub owner/repo")
+    parser.add_argument("--source-run-id", type=int, help="Source workflow run id")
+    parser.add_argument("--artifact-name", help="Artifact name in source run")
+    parser.add_argument("--github-token", help="GitHub token for artifact download")
+    parser.add_argument(
+        "--dify-base-url",
+        default="",
+        help="Dify service API base URL (or use KNOWLEDGE_DIFY_BASE_URL)",
+    )
+    parser.add_argument(
+        "--dify-api-key",
+        default="",
+        help="Dify service API key (or use KNOWLEDGE_DIFY_API_KEY)",
+    )
+    parser.add_argument(
+        "--skip-dify",
+        action="store_true",
+        help="Skip live Dify publication and only write Git state",
+    )
+    parser.add_argument(
+        "--summary-json",
+        type=Path,
+        help="Optional summary output path",
+    )
+    return parser.parse_args()
+
+
+def resolve_artifact_zip(args: argparse.Namespace) -> Path:
+    """Resolve the artifact zip either locally or from GitHub Actions."""
+
+    if args.artifact_file:
+        return args.artifact_file.resolve()
+
+    source_repo = require_non_empty_str(args.source_repo, "--source-repo")
+    artifact_name = require_non_empty_str(args.artifact_name, "--artifact-name")
+    github_token = require_non_empty_str(args.github_token, "--github-token")
+    if not isinstance(args.source_run_id, int) or args.source_run_id <= 0:
+        raise KnowledgeContractError("--source-run-id 必须是正整数")
+
+    temp_root = Path(tempfile.mkdtemp(prefix="knowledge-ingest-"))
+    artifact_path = temp_root / f"{artifact_name}.zip"
+    try:
+        return download_artifact_zip(
+            repository=source_repo,
+            run_id=args.source_run_id,
+            artifact_name=artifact_name,
+            token=github_token,
+            output_path=artifact_path,
+        )
+    except GitHubArtifactError as exc:
+        shutil.rmtree(temp_root, ignore_errors=True)
+        raise KnowledgeContractError(str(exc)) from exc
+
+
+def extract_artifact(artifact_zip: Path) -> Path:
+    """Extract an artifact zip into a temporary directory."""
+
+    if not artifact_zip.exists():
+        raise KnowledgeContractError(f"artifact zip 不存在: {artifact_zip}")
+    extract_root = Path(tempfile.mkdtemp(prefix="knowledge-artifact-"))
+    with zipfile.ZipFile(artifact_zip) as archive:
+        archive.extractall(extract_root)
+    return extract_root
+
+
+def validate_source_registration(
+    *,
+    deploy_root: Path,
+    source_repo: str,
+    service_name: str,
+    documents: list[JSONMapping],
+) -> None:
+    """Ensure the source repo and source ids are registered centrally."""
+
+    registry = load_source_registry(deploy_root)
+    entry = registry.get(source_repo)
+    if entry is None:
+        raise KnowledgeContractError(
+            f"source repo 未在中央 registry 注册: {source_repo}"
+        )
+    registered_service = require_non_empty_str(entry.get("service"), "registry.service")
+    if registered_service != service_name:
+        raise KnowledgeContractError(
+            f"source repo service 不匹配: {source_repo} -> {registered_service} != {service_name}"
+        )
+    registered_source_ids = {
+        require_non_empty_str(item, "registry.source_ids[]")
+        for item in cast(list[str], entry.get("source_ids", []))
+    }
+    allowed_visibilities = {
+        require_non_empty_str(item, "registry.allowed_visibilities[]")
+        for item in cast(list[str], entry.get("allowed_visibilities", []))
+    }
+    for index, document in enumerate(documents):
+        source_id = require_non_empty_str(
+            document.get("source_id"), f"artifact.documents[{index}].source_id"
+        )
+        if source_id not in registered_source_ids:
+            raise KnowledgeContractError(
+                f"source_id 未在中央 registry 注册: {source_repo}/{source_id}"
+            )
+        visibility = optional_non_empty_str(
+            document.get("visibility"), f"artifact.documents[{index}].visibility"
+        )
+        if visibility and visibility not in allowed_visibilities:
+            raise KnowledgeContractError(
+                f"visibility 不在 source repo 白名单中: {source_repo}/{visibility}"
+            )
+
+
+def load_artifact_manifest(extract_root: Path, deploy_root: Path) -> JSONMapping:
+    """Load and validate manifest.json from an extracted artifact."""
+
+    manifest_path = extract_root / "manifest.json"
+    if not manifest_path.exists():
+        raise KnowledgeContractError(f"artifact 缺少 manifest.json: {extract_root}")
+    manifest = load_json_mapping(manifest_path)
+    validator = load_validator(deploy_root / ARTIFACT_SCHEMA_RELATIVE)
+    validate_payload(validator, manifest, "knowledge artifact manifest")
+    return manifest
+
+
+def copy_artifact_snapshot(
+    *,
+    extract_root: Path,
+    deploy_root: Path,
+    environment: str,
+    source_repo: str,
+    artifact_id: str,
+) -> Path:
+    """Persist the raw artifact payload into ljwx-deploy."""
+
+    snapshot_root = artifact_snapshot_root(
+        deploy_root, environment, source_repo, artifact_id
+    )
+    if snapshot_root.exists():
+        shutil.rmtree(snapshot_root)
+    snapshot_root.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copytree(extract_root, snapshot_root)
+    return snapshot_root
+
+
+def load_artifact_content(snapshot_root: Path, document_entry: JSONMapping) -> str:
+    """Load one markdown content file from the snapshot."""
+
+    content_path = optional_non_empty_str(
+        document_entry.get("content_path"), "content_path"
+    )
+    if content_path is None:
+        raise KnowledgeContractError("upsert 文档缺少 content_path")
+    absolute_path = snapshot_root / content_path
+    if not absolute_path.exists():
+        raise KnowledgeContractError(f"artifact content 不存在: {absolute_path}")
+    return absolute_path.read_text(encoding="utf-8")
+
+
+def build_normalized_document(
+    document_entry: JSONMapping,
+    acl_policies: dict[str, dict[str, object]],
+) -> JSONMapping:
+    """Build a normalized document contract row."""
+
+    visibility = require_non_empty_str(
+        document_entry.get("visibility"), "document.visibility"
+    )
+    normalized: JSONMapping = {
+        "document_id": require_non_empty_str(
+            document_entry.get("document_id"), "document.document_id"
+        ),
+        "title": require_non_empty_str(document_entry.get("title"), "document.title"),
+        "source_type": require_non_empty_str(
+            document_entry.get("source_type"), "document.source_type"
+        ),
+        "source_ref": require_non_empty_str(
+            document_entry.get("source_ref"), "document.source_ref"
+        ),
+        "product_area": require_non_empty_str(
+            document_entry.get("product_area"), "document.product_area"
+        ),
+        "visibility": visibility,
+        "acl_expression": build_acl_expression(visibility, acl_policies),
+        "customer_id": None,
+        "doc_type": require_non_empty_str(
+            document_entry.get("doc_type"), "document.doc_type"
+        ),
+        "owner": require_non_empty_str(document_entry.get("owner"), "document.owner"),
+        "version": require_non_empty_str(
+            document_entry.get("version"), "document.version"
+        ),
+        "checksum": require_non_empty_str(
+            document_entry.get("checksum"), "document.checksum"
+        ),
+        "review_status": require_non_empty_str(
+            document_entry.get("review_status"), "document.review_status"
+        ),
+        "approved_by": optional_non_empty_str(
+            document_entry.get("approved_by"), "document.approved_by"
+        ),
+        "effective_from": optional_non_empty_str(
+            document_entry.get("effective_from"), "document.effective_from"
+        ),
+        "expires_at": optional_non_empty_str(
+            document_entry.get("expires_at"), "document.expires_at"
+        ),
+        "tags": list(cast(list[str], document_entry.get("tags", []))),
+    }
+    return normalized
+
+
+def build_chunk_rows(
+    normalized_document: JSONMapping, content: str
+) -> list[JSONMapping]:
+    """Build chunk contract rows for one normalized document."""
+
+    document_id = require_non_empty_str(
+        normalized_document.get("document_id"), "normalized.document_id"
+    )
+    visibility = require_non_empty_str(
+        normalized_document.get("visibility"), "normalized.visibility"
+    )
+    acl_expression = require_non_empty_str(
+        normalized_document.get("acl_expression"), "normalized.acl_expression"
+    )
+    chunks: list[JSONMapping] = []
+    for index, chunk_text in enumerate(chunk_markdown(content)):
+        chunks.append(
+            {
+                "chunk_id": f"{document_id}:chunk:{index:04d}",
+                "document_id": document_id,
+                "chunk_index": index,
+                "heading_path": [],
+                "visibility": visibility,
+                "acl_expression": acl_expression,
+                "customer_id": None,
+                "token_count": max(1, len(chunk_text.strip()) // 4),
+                "embedding_model": None,
+                "text_hash": sha256_text(chunk_text),
+                "published_targets": [],
+            }
+        )
+    return chunks
+
+
+def validate_contract_rows(
+    *,
+    deploy_root: Path,
+    normalized_rows: list[JSONMapping],
+    chunk_rows: list[JSONMapping],
+) -> None:
+    """Validate normalized document and chunk rows."""
+
+    document_validator = load_validator(deploy_root / DOCUMENT_SCHEMA_RELATIVE)
+    chunk_validator = load_validator(deploy_root / CHUNK_SCHEMA_RELATIVE)
+    for row in normalized_rows:
+        document_id = require_non_empty_str(
+            row.get("document_id"), "normalized.document_id"
+        )
+        validate_payload(document_validator, row, f"normalized<{document_id}>")
+    for row in chunk_rows:
+        chunk_id = require_non_empty_str(row.get("chunk_id"), "chunk.chunk_id")
+        validate_payload(chunk_validator, row, f"chunk<{chunk_id}>")
+
+
+def build_publishable_target_ids(
+    *,
+    normalized_document: JSONMapping,
+    acl_policies: dict[str, dict[str, object]],
+    publish_targets: dict[str, dict[str, object]],
+) -> list[str]:
+    """Resolve the Dify targets allowed for a normalized document."""
+
+    visibility = require_non_empty_str(
+        normalized_document.get("visibility"), "visibility"
+    )
+    policy = acl_policies.get(visibility)
+    if policy is None:
+        raise KnowledgeContractError(f"未找到 ACL policy: {visibility}")
+    allowed_targets = cast(list[str], policy.get("allowed_targets", []))
+
+    resolved: list[str] = []
+    for target_id in allowed_targets:
+        target = publish_targets.get(target_id)
+        if target is None:
+            continue
+        target_type = require_non_empty_str(
+            target.get("type"), f"target.{target_id}.type"
+        )
+        target_visibility = require_non_empty_str(
+            target.get("visibility"), f"target.{target_id}.visibility"
+        )
+        if target_type != "dify_dataset":
+            continue
+        if target_visibility != visibility:
+            continue
+        resolved.append(target_id)
+    return resolved
+
+
+def remove_remote_targets(
+    *,
+    record: JSONMapping,
+    client: DifyClient | None,
+    skip_dify: bool,
+) -> list[JSONMapping]:
+    """Remove remote documents recorded on a publish record."""
+
+    updated_targets: list[JSONMapping] = []
+    raw_targets = cast_mapping_list(record.get("targets", []), "record.targets")
+    now = utc_now()
+    for target in raw_targets:
+        dataset_id = optional_non_empty_str(
+            target.get("dataset_id"), "target.dataset_id"
+        )
+        remote_document_id = optional_non_empty_str(
+            target.get("remote_document_id"), "target.remote_document_id"
+        )
+        if not skip_dify and client and dataset_id and remote_document_id:
+            client.delete_document(dataset_id, remote_document_id)
+        target["status"] = "invalidated"
+        target["updated_at"] = now
+        updated_targets.append(target)
+    return updated_targets
+
+
+def choose_duplicate_of(
+    *,
+    records: dict[str, JSONMapping],
+    document_id: str,
+    visibility: str,
+    checksum: str,
+) -> str | None:
+    """Find an already-published record with the same checksum."""
+
+    for existing_document_id, record in records.items():
+        if existing_document_id == document_id:
+            continue
+        if optional_non_empty_str(record.get("status"), "record.status") != "published":
+            continue
+        if (
+            optional_non_empty_str(record.get("visibility"), "record.visibility")
+            != visibility
+        ):
+            continue
+        if (
+            optional_non_empty_str(record.get("checksum"), "record.checksum")
+            != checksum
+        ):
+            continue
+        return existing_document_id
+    return None
+
+
+def publish_targets_to_dify(
+    *,
+    client: DifyClient | None,
+    skip_dify: bool,
+    target_ids: list[str],
+    publish_targets: dict[str, dict[str, object]],
+    document_name: str,
+    content: str,
+    existing_record: JSONMapping | None,
+) -> list[JSONMapping]:
+    """Publish one document to all resolved Dify targets."""
+
+    existing_target_by_id: dict[str, JSONMapping] = {}
+    if existing_record is not None:
+        for item in cast_mapping_list(
+            existing_record.get("targets", []), "record.targets"
+        ):
+            target_id = require_non_empty_str(
+                item.get("target_id"), "record.targets[].target_id"
+            )
+            existing_target_by_id[target_id] = item
+
+    results: list[JSONMapping] = []
+    now = utc_now()
+    for target_id in target_ids:
+        target = publish_targets[target_id]
+        dataset_name = require_non_empty_str(
+            target.get("dataset_name"), f"publish-targets.{target_id}.dataset_name"
+        )
+        dataset_id: str | None = None
+        remote_document_id: str | None = None
+        existing_target = existing_target_by_id.get(target_id)
+        if existing_target is not None:
+            dataset_id = optional_non_empty_str(
+                existing_target.get("dataset_id"), "existing_target.dataset_id"
+            )
+            remote_document_id = optional_non_empty_str(
+                existing_target.get("remote_document_id"),
+                "existing_target.remote_document_id",
+            )
+
+        if not skip_dify:
+            if client is None:
+                raise KnowledgeContractError("Dify client 未初始化")
+            dataset = client.ensure_dataset(dataset_name)
+            dataset_id = dataset.dataset_id
+            if remote_document_id is None:
+                remote_document_id = client.find_document_id(dataset_id, document_name)
+            remote_document_id = client.upsert_document(
+                dataset_id=dataset_id,
+                document_name=document_name,
+                content=content,
+                document_id=remote_document_id,
+            )
+
+        results.append(
+            {
+                "target_id": target_id,
+                "target_type": require_non_empty_str(
+                    target.get("type"), f"target.{target_id}.type"
+                ),
+                "dataset_name": dataset_name,
+                "dataset_id": dataset_id,
+                "document_name": document_name,
+                "remote_document_id": remote_document_id,
+                "status": "published" if not skip_dify else "skipped",
+                "updated_at": now,
+            }
+        )
+    return results
+
+
+def maybe_build_dify_client(args: argparse.Namespace) -> DifyClient | None:
+    """Create a Dify client when live publication is enabled."""
+
+    if args.skip_dify:
+        return None
+    base_url = require_non_empty_str(args.dify_base_url, "--dify-base-url")
+    api_key = require_non_empty_str(args.dify_api_key, "--dify-api-key")
+    try:
+        return DifyClient(base_url=base_url, api_key=api_key)
+    except DifyClientError as exc:
+        raise KnowledgeContractError(str(exc)) from exc
+
+
+def main() -> int:
+    """CLI entry point."""
+
+    args = parse_args()
+    deploy_root = args.deploy_root.resolve()
+    client = maybe_build_dify_client(args)
+    artifact_zip = resolve_artifact_zip(args)
+    extract_root = extract_artifact(artifact_zip)
+
+    try:
+        manifest = load_artifact_manifest(extract_root, deploy_root)
+        artifact_id = require_non_empty_str(
+            manifest.get("artifact_id"), "manifest.artifact_id"
+        )
+        source = require_mapping(manifest.get("source"), "manifest.source")
+        source_repo = require_non_empty_str(source.get("repo"), "manifest.source.repo")
+        service_name = require_non_empty_str(
+            source.get("service"), "manifest.source.service"
+        )
+        raw_documents = cast_mapping_list(
+            manifest.get("documents"), "manifest.documents"
+        )
+        validate_source_registration(
+            deploy_root=deploy_root,
+            source_repo=source_repo,
+            service_name=service_name,
+            documents=raw_documents,
+        )
+
+        snapshot_root = copy_artifact_snapshot(
+            extract_root=extract_root,
+            deploy_root=deploy_root,
+            environment=args.env,
+            source_repo=source_repo,
+            artifact_id=artifact_id,
+        )
+        artifact_ref = (
+            (snapshot_root / "manifest.json").relative_to(deploy_root).as_posix()
+        )
+        normalized_path = snapshot_root / "normalized-documents.jsonl"
+        classified_path = snapshot_root / "classified-documents.jsonl"
+        chunks_path = snapshot_root / "chunks.jsonl"
+
+        acl_policies = load_acl_policies(deploy_root)
+        publish_targets = load_publish_targets(deploy_root)
+        publish_record_validator = load_validator(
+            deploy_root / PUBLISH_RECORD_SCHEMA_RELATIVE
+        )
+        records_root = knowledge_paths(deploy_root, args.env).publish_records_root
+        records = load_publish_records(records_root)
+
+        normalized_rows: list[JSONMapping] = []
+        classified_rows: list[JSONMapping] = []
+        all_chunk_rows: list[JSONMapping] = []
+        events: list[JSONMapping] = []
+        published_documents = 0
+        invalidated_documents = 0
+
+        invalidate_entries = [
+            item
+            for item in raw_documents
+            if require_non_empty_str(item.get("op"), "document.op") == "invalidate"
+        ]
+        upsert_entries = [
+            item
+            for item in raw_documents
+            if require_non_empty_str(item.get("op"), "document.op") == "upsert"
+        ]
+
+        for document_entry in invalidate_entries:
+            document_id = require_non_empty_str(
+                document_entry.get("document_id"), "document.document_id"
+            )
+            existing_record = records.get(document_id)
+            if existing_record is None:
+                events.append(
+                    build_event(
+                        action="invalidate",
+                        artifact_id=artifact_id,
+                        document_id=document_id,
+                        source_repo=source_repo,
+                        status="ignored",
+                        reason="source_deleted_without_existing_record",
+                        target_ids=[],
+                    )
+                )
+                continue
+            updated_targets = remove_remote_targets(
+                record=existing_record,
+                client=client,
+                skip_dify=args.skip_dify,
+            )
+            existing_record["status"] = "invalidated"
+            existing_record["latest_action"] = "invalidate"
+            existing_record["latest_reason"] = require_non_empty_str(
+                document_entry.get("reason"), "document.reason"
+            )
+            existing_record["updated_at"] = utc_now()
+            existing_record["targets"] = updated_targets
+            validate_payload(
+                publish_record_validator,
+                existing_record,
+                f"publish-record<{document_id}>",
+            )
+            record_path = record_path_for_document(
+                deploy_root, args.env, source_repo, document_id
+            )
+            write_json(record_path, existing_record)
+            records[document_id] = existing_record
+            invalidated_documents += 1
+            events.append(
+                build_event(
+                    action="invalidate",
+                    artifact_id=artifact_id,
+                    document_id=document_id,
+                    source_repo=source_repo,
+                    status="invalidated",
+                    reason=require_non_empty_str(
+                        document_entry.get("reason"), "document.reason"
+                    ),
+                    target_ids=[
+                        require_non_empty_str(item.get("target_id"), "target.target_id")
+                        for item in updated_targets
+                    ],
+                )
+            )
+
+        for document_entry in upsert_entries:
+            document_id = require_non_empty_str(
+                document_entry.get("document_id"), "document.document_id"
+            )
+            existing_record = records.get(document_id)
+            content = load_artifact_content(snapshot_root, document_entry)
+            normalized_document = build_normalized_document(
+                document_entry, acl_policies
+            )
+            normalized_rows.append(normalized_document)
+            classified_rows.append(dict(normalized_document))
+            chunk_rows = build_chunk_rows(normalized_document, content)
+            all_chunk_rows.extend(chunk_rows)
+
+            visibility = require_non_empty_str(
+                normalized_document.get("visibility"), "normalized.visibility"
+            )
+            checksum = require_non_empty_str(
+                normalized_document.get("checksum"), "normalized.checksum"
+            )
+            review_status = require_non_empty_str(
+                normalized_document.get("review_status"), "normalized.review_status"
+            )
+            publish_target_ids = build_publishable_target_ids(
+                normalized_document=normalized_document,
+                acl_policies=acl_policies,
+                publish_targets=publish_targets,
+            )
+            duplicate_of = choose_duplicate_of(
+                records=records,
+                document_id=document_id,
+                visibility=visibility,
+                checksum=checksum,
+            )
+            content_ref = (
+                (
+                    snapshot_root
+                    / require_non_empty_str(
+                        document_entry.get("content_path"), "document.content_path"
+                    )
+                )
+                .relative_to(deploy_root)
+                .as_posix()
+            )
+            base_record: JSONMapping = {
+                "schema_version": "0.1.0",
+                "document_id": document_id,
+                "source_repo": source_repo,
+                "source_id": require_non_empty_str(
+                    document_entry.get("source_id"), "document.source_id"
+                ),
+                "source_path": require_non_empty_str(
+                    document_entry.get("source_path"), "document.source_path"
+                ),
+                "source_ref": require_non_empty_str(
+                    document_entry.get("source_ref"), "document.source_ref"
+                ),
+                "visibility": visibility,
+                "checksum": checksum,
+                "title": require_non_empty_str(
+                    document_entry.get("title"), "document.title"
+                ),
+                "source_type": require_non_empty_str(
+                    document_entry.get("source_type"), "document.source_type"
+                ),
+                "doc_type": require_non_empty_str(
+                    document_entry.get("doc_type"), "document.doc_type"
+                ),
+                "product_area": require_non_empty_str(
+                    document_entry.get("product_area"), "document.product_area"
+                ),
+                "owner": require_non_empty_str(
+                    document_entry.get("owner"), "document.owner"
+                ),
+                "review_status": review_status,
+                "acl_expression": require_non_empty_str(
+                    normalized_document.get("acl_expression"),
+                    "normalized.acl_expression",
+                ),
+                "version": require_non_empty_str(
+                    document_entry.get("version"), "document.version"
+                ),
+                "tags": list(cast(list[str], document_entry.get("tags", []))),
+                "status": "published",
+                "duplicate_of": None,
+                "artifact_ref": artifact_ref,
+                "content_ref": content_ref,
+                "normalized_ref": normalized_path.relative_to(deploy_root).as_posix(),
+                "classified_ref": classified_path.relative_to(deploy_root).as_posix(),
+                "chunk_ref": chunks_path.relative_to(deploy_root).as_posix(),
+                "chunk_count": len(chunk_rows),
+                "latest_action": "publish",
+                "latest_reason": "artifact_ingest",
+                "published_at": utc_now(),
+                "updated_at": utc_now(),
+                "targets": [],
+            }
+
+            if review_status != "approved":
+                if existing_record is not None:
+                    base_record["targets"] = remove_remote_targets(
+                        record=existing_record,
+                        client=client,
+                        skip_dify=args.skip_dify,
+                    )
+                base_record["status"] = "invalidated"
+                base_record["latest_action"] = "invalidate"
+                base_record["latest_reason"] = f"review_status={review_status}"
+                base_record["published_at"] = None
+                invalidated_documents += 1
+                events.append(
+                    build_event(
+                        action="invalidate",
+                        artifact_id=artifact_id,
+                        document_id=document_id,
+                        source_repo=source_repo,
+                        status="invalidated",
+                        reason=f"review_status={review_status}",
+                        target_ids=[],
+                    )
+                )
+            elif duplicate_of is not None:
+                if existing_record is not None:
+                    base_record["targets"] = remove_remote_targets(
+                        record=existing_record,
+                        client=client,
+                        skip_dify=args.skip_dify,
+                    )
+                base_record["status"] = "deduplicated"
+                base_record["duplicate_of"] = duplicate_of
+                base_record["latest_action"] = "invalidate"
+                base_record["latest_reason"] = f"duplicate_of={duplicate_of}"
+                base_record["published_at"] = None
+                invalidated_documents += 1
+                events.append(
+                    build_event(
+                        action="invalidate",
+                        artifact_id=artifact_id,
+                        document_id=document_id,
+                        source_repo=source_repo,
+                        status="deduplicated",
+                        reason=f"duplicate_of={duplicate_of}",
+                        target_ids=[],
+                    )
+                )
+            else:
+                if existing_record is not None:
+                    previous_targets = {
+                        require_non_empty_str(item.get("target_id"), "record.target_id")
+                        for item in cast_mapping_list(
+                            existing_record.get("targets", []), "record.targets"
+                        )
+                    }
+                    if set(publish_target_ids) != previous_targets:
+                        remove_remote_targets(
+                            record=existing_record,
+                            client=client,
+                            skip_dify=args.skip_dify,
+                        )
+                try:
+                    base_record["targets"] = publish_targets_to_dify(
+                        client=client,
+                        skip_dify=args.skip_dify,
+                        target_ids=publish_target_ids,
+                        publish_targets=publish_targets,
+                        document_name=artifact_document_name(document_id),
+                        content=content,
+                        existing_record=existing_record,
+                    )
+                except DifyClientError as exc:
+                    raise KnowledgeContractError(str(exc)) from exc
+                published_documents += 1
+                events.append(
+                    build_event(
+                        action="publish",
+                        artifact_id=artifact_id,
+                        document_id=document_id,
+                        source_repo=source_repo,
+                        status="published" if not args.skip_dify else "staged",
+                        reason="artifact_ingest",
+                        target_ids=[
+                            require_non_empty_str(
+                                item.get("target_id"), "target.target_id"
+                            )
+                            for item in cast_mapping_list(
+                                base_record.get("targets", []), "record.targets"
+                            )
+                        ],
+                    )
+                )
+
+            validate_payload(
+                publish_record_validator,
+                base_record,
+                f"publish-record<{document_id}>",
+            )
+            record_path = record_path_for_document(
+                deploy_root, args.env, source_repo, document_id
+            )
+            write_json(record_path, base_record)
+            records[document_id] = base_record
+
+        validate_contract_rows(
+            deploy_root=deploy_root,
+            normalized_rows=normalized_rows,
+            chunk_rows=all_chunk_rows,
+        )
+        write_jsonl(normalized_path, normalized_rows)
+        write_jsonl(classified_path, classified_rows)
+        write_jsonl(chunks_path, all_chunk_rows)
+        for event in events:
+            append_jsonl(knowledge_paths(deploy_root, args.env).events_path, event)
+
+        summary: JSONMapping = {
+            "artifact_id": artifact_id,
+            "environment": args.env,
+            "published_documents": published_documents,
+            "invalidated_documents": invalidated_documents,
+            "normalized_documents": len(normalized_rows),
+            "classified_documents": len(classified_rows),
+            "chunk_rows": len(all_chunk_rows),
+            "snapshot_root": snapshot_root.relative_to(deploy_root).as_posix(),
+            "source_repo": source_repo,
+        }
+        if args.summary_json:
+            write_json(args.summary_json.resolve(), summary)
+        print(json.dumps(summary, ensure_ascii=False))
+        return 0
+    finally:
+        artifact_temp_root = artifact_zip.parent
+        if artifact_temp_root.name.startswith("knowledge-ingest-"):
+            shutil.rmtree(artifact_temp_root, ignore_errors=True)
+        shutil.rmtree(extract_root, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/knowledge/reconcile_records.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Replay published knowledge records into Dify dev targets."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from .common import (
+    JSONMapping,
+    KnowledgeContractError,
+    PUBLISH_RECORD_SCHEMA_RELATIVE,
+    append_jsonl,
+    artifact_document_name,
+    build_event,
+    cast_mapping_list,
+    knowledge_paths,
+    load_publish_records,
+    load_validator,
+    optional_non_empty_str,
+    require_non_empty_str,
+    utc_now,
+    validate_payload,
+    write_json,
+)
+from .dify_client import DifyClient, DifyClientError
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse CLI arguments."""
+
+    parser = argparse.ArgumentParser(
+        description="Reconcile published knowledge records"
+    )
+    parser.add_argument("--deploy-root", type=Path, required=True)
+    parser.add_argument("--env", default="dev")
+    parser.add_argument("--source-repo")
+    parser.add_argument("--document-id")
+    parser.add_argument("--dify-base-url", default="")
+    parser.add_argument("--dify-api-key", default="")
+    parser.add_argument("--skip-dify", action="store_true")
+    parser.add_argument("--summary-json", type=Path)
+    return parser.parse_args()
+
+
+def maybe_build_dify_client(args: argparse.Namespace) -> DifyClient | None:
+    """Create a Dify client when live reconciliation is enabled."""
+
+    if args.skip_dify:
+        return None
+    try:
+        return DifyClient(
+            base_url=require_non_empty_str(args.dify_base_url, "--dify-base-url"),
+            api_key=require_non_empty_str(args.dify_api_key, "--dify-api-key"),
+        )
+    except DifyClientError as exc:
+        raise KnowledgeContractError(str(exc)) from exc
+
+
+def main() -> int:
+    """CLI entry point."""
+
+    args = parse_args()
+    deploy_root = args.deploy_root.resolve()
+    client = maybe_build_dify_client(args)
+    records = load_publish_records(
+        knowledge_paths(deploy_root, args.env).publish_records_root
+    )
+    validator = load_validator(deploy_root / PUBLISH_RECORD_SCHEMA_RELATIVE)
+
+    reconciled = 0
+    for document_id, record in records.items():
+        source_repo = require_non_empty_str(
+            record.get("source_repo"), "record.source_repo"
+        )
+        if args.source_repo and source_repo != args.source_repo:
+            continue
+        if args.document_id and document_id != args.document_id:
+            continue
+        status = require_non_empty_str(record.get("status"), "record.status")
+        if status != "published":
+            continue
+        content_ref = optional_non_empty_str(
+            record.get("content_ref"), "record.content_ref"
+        )
+        if content_ref is None:
+            continue
+        content_path = deploy_root / content_ref
+        if not content_path.exists():
+            raise KnowledgeContractError(f"content_ref 不存在: {content_path}")
+        content = content_path.read_text(encoding="utf-8")
+
+        updated_targets: list[JSONMapping] = []
+        for target in cast_mapping_list(record.get("targets", []), "record.targets"):
+            target_id = require_non_empty_str(
+                target.get("target_id"), "target.target_id"
+            )
+            dataset_name = require_non_empty_str(
+                target.get("dataset_name"), "target.dataset_name"
+            )
+            dataset_id = optional_non_empty_str(
+                target.get("dataset_id"), "target.dataset_id"
+            )
+            remote_document_id = optional_non_empty_str(
+                target.get("remote_document_id"), "target.remote_document_id"
+            )
+            if not args.skip_dify:
+                if client is None:
+                    raise KnowledgeContractError("Dify client 未初始化")
+                dataset = client.ensure_dataset(dataset_name)
+                dataset_id = dataset.dataset_id
+                if remote_document_id is None:
+                    remote_document_id = client.find_document_id(
+                        dataset.dataset_id, artifact_document_name(document_id)
+                    )
+                remote_document_id = client.upsert_document(
+                    dataset_id=dataset.dataset_id,
+                    document_name=artifact_document_name(document_id),
+                    content=content,
+                    document_id=remote_document_id,
+                )
+            updated_targets.append(
+                {
+                    "target_id": target_id,
+                    "target_type": require_non_empty_str(
+                        target.get("target_type"), "target.target_type"
+                    ),
+                    "dataset_name": dataset_name,
+                    "dataset_id": dataset_id,
+                    "document_name": artifact_document_name(document_id),
+                    "remote_document_id": remote_document_id,
+                    "status": "published" if not args.skip_dify else "skipped",
+                    "updated_at": utc_now(),
+                }
+            )
+
+        record["targets"] = updated_targets
+        record["latest_action"] = "republish"
+        record["latest_reason"] = "manual_reconcile"
+        record["updated_at"] = utc_now()
+        validate_payload(validator, record, f"publish-record<{document_id}>")
+        source_repo_slug = source_repo.replace("/", "__")
+        record_path = (
+            knowledge_paths(deploy_root, args.env).publish_records_root
+            / source_repo_slug
+            / f"{document_id}.json"
+        )
+        write_json(record_path, record)
+        append_jsonl(
+            knowledge_paths(deploy_root, args.env).events_path,
+            build_event(
+                action="republish",
+                artifact_id=require_non_empty_str(
+                    record.get("artifact_ref"), "record.artifact_ref"
+                ),
+                document_id=document_id,
+                source_repo=source_repo,
+                status="published" if not args.skip_dify else "staged",
+                reason="manual_reconcile",
+                target_ids=[
+                    require_non_empty_str(item.get("target_id"), "target.target_id")
+                    for item in updated_targets
+                ],
+            ),
+        )
+        reconciled += 1
+
+    summary: JSONMapping = {
+        "environment": args.env,
+        "document_id": args.document_id or "",
+        "source_repo": args.source_repo or "",
+        "reconciled_records": reconciled,
+    }
+    if args.summary_json:
+        write_json(args.summary_json.resolve(), summary)
+    print(json.dumps(summary, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/knowledge/unpublish_records.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""Manually unpublish one knowledge document from central targets."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from .common import (
+    JSONMapping,
+    KnowledgeContractError,
+    PUBLISH_RECORD_SCHEMA_RELATIVE,
+    append_jsonl,
+    build_event,
+    cast_mapping_list,
+    knowledge_paths,
+    load_validator,
+    record_path_for_document,
+    require_non_empty_str,
+    utc_now,
+    validate_payload,
+    write_json,
+)
+from .dify_client import DifyClient, DifyClientError
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse CLI arguments."""
+
+    parser = argparse.ArgumentParser(description="Unpublish one knowledge document")
+    parser.add_argument("--deploy-root", type=Path, required=True)
+    parser.add_argument("--env", default="dev")
+    parser.add_argument("--source-repo", required=True)
+    parser.add_argument("--document-id", required=True)
+    parser.add_argument("--reason", default="manual_unpublish")
+    parser.add_argument("--dify-base-url", default="")
+    parser.add_argument("--dify-api-key", default="")
+    parser.add_argument("--skip-dify", action="store_true")
+    parser.add_argument("--summary-json", type=Path)
+    return parser.parse_args()
+
+
+def maybe_build_dify_client(args: argparse.Namespace) -> DifyClient | None:
+    """Create a Dify client when live publication is enabled."""
+
+    if args.skip_dify:
+        return None
+    try:
+        return DifyClient(
+            base_url=require_non_empty_str(args.dify_base_url, "--dify-base-url"),
+            api_key=require_non_empty_str(args.dify_api_key, "--dify-api-key"),
+        )
+    except DifyClientError as exc:
+        raise KnowledgeContractError(str(exc)) from exc
+
+
+def main() -> int:
+    """CLI entry point."""
+
+    args = parse_args()
+    deploy_root = args.deploy_root.resolve()
+    client = maybe_build_dify_client(args)
+    record_path = record_path_for_document(
+        deploy_root, args.env, args.source_repo, args.document_id
+    )
+    if not record_path.exists():
+        raise KnowledgeContractError(f"未找到 publish record: {record_path}")
+
+    record = json.loads(record_path.read_text(encoding="utf-8"))
+    if not isinstance(record, dict):
+        raise KnowledgeContractError(
+            f"publish record 根节点必须是 object: {record_path}"
+        )
+    payload = {str(key): value for key, value in record.items()}
+
+    updated_targets: list[JSONMapping] = []
+    for target in cast_mapping_list(payload.get("targets", []), "record.targets"):
+        dataset_id = target.get("dataset_id")
+        remote_document_id = target.get("remote_document_id")
+        if (
+            not args.skip_dify
+            and client is not None
+            and isinstance(dataset_id, str)
+            and dataset_id
+            and isinstance(remote_document_id, str)
+            and remote_document_id
+        ):
+            client.delete_document(dataset_id, remote_document_id)
+        target["status"] = "unpublished"
+        target["updated_at"] = utc_now()
+        updated_targets.append(target)
+
+    payload["status"] = "unpublished"
+    payload["latest_action"] = "unpublish"
+    payload["latest_reason"] = args.reason
+    payload["updated_at"] = utc_now()
+    payload["targets"] = updated_targets
+
+    validator = load_validator(deploy_root / PUBLISH_RECORD_SCHEMA_RELATIVE)
+    validate_payload(validator, payload, f"publish-record<{args.document_id}>")
+    write_json(record_path, payload)
+    append_jsonl(
+        knowledge_paths(deploy_root, args.env).events_path,
+        build_event(
+            action="unpublish",
+            artifact_id=require_non_empty_str(
+                payload.get("artifact_ref"), "record.artifact_ref"
+            ),
+            document_id=args.document_id,
+            source_repo=args.source_repo,
+            status="unpublished",
+            reason=args.reason,
+            target_ids=[
+                require_non_empty_str(item.get("target_id"), "target.target_id")
+                for item in updated_targets
+            ],
+        ),
+    )
+
+    summary: JSONMapping = {
+        "document_id": args.document_id,
+        "environment": args.env,
+        "source_repo": args.source_repo,
+        "targets": len(updated_targets),
+        "status": "unpublished",
+    }
+    if args.summary_json:
+        write_json(args.summary_json.resolve(), summary)
+    print(json.dumps(summary, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/knowledge/validate_state.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""Validate committed knowledge publish records."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+try:
+    import yaml
+except ImportError as exc:
+    raise SystemExit(
+        f"缺少依赖 PyYAML，请使用 uvx --with pyyaml --with jsonschema python <script>\n{exc}"
+    ) from exc
+
+from .common import (
+    JSONMapping,
+    KnowledgeContractError,
+    PUBLISH_RECORD_SCHEMA_RELATIVE,
+    knowledge_paths,
+    load_publish_records,
+    load_validator,
+    optional_non_empty_str,
+    require_non_empty_str,
+    validate_payload,
+)
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse CLI arguments."""
+
+    parser = argparse.ArgumentParser(description="Validate knowledge publish records")
+    parser.add_argument(
+        "--deploy-root", type=Path, default=Path("."), help="ljwx-deploy root"
+    )
+    parser.add_argument("--env", default="dev")
+    return parser.parse_args()
+
+
+def load_platform_version(path: Path) -> JSONMapping:
+    """Load release/platform-version.yaml."""
+
+    raw = yaml.safe_load(path.read_text(encoding="utf-8"))
+    if not isinstance(raw, dict):
+        raise KnowledgeContractError(f"{path} 根节点必须是 mapping")
+    return {str(key): value for key, value in raw.items()}
+
+
+def main() -> int:
+    """CLI entry point."""
+
+    args = parse_args()
+    deploy_root = args.deploy_root.resolve()
+    validator = load_validator(deploy_root / PUBLISH_RECORD_SCHEMA_RELATIVE)
+    records = load_publish_records(
+        knowledge_paths(deploy_root, args.env).publish_records_root
+    )
+    release_version = load_platform_version(
+        deploy_root / "release/platform-version.yaml"
+    )
+    config_versions = release_version.get("config_versions")
+    if not isinstance(config_versions, dict):
+        raise KnowledgeContractError(
+            "release/platform-version.yaml.config_versions 缺失"
+        )
+    publish_record_ref = optional_non_empty_str(
+        config_versions.get("knowledge_publish_record_schema"),
+        "release/platform-version.yaml.config_versions.knowledge_publish_record_schema",
+    )
+    expected_ref = f"{PUBLISH_RECORD_SCHEMA_RELATIVE.as_posix()}@0.1.0"
+    if publish_record_ref != expected_ref:
+        raise KnowledgeContractError(
+            "release/platform-version.yaml 中 knowledge_publish_record_schema "
+            f"必须等于 {expected_ref}"
+        )
+
+    for document_id, record in records.items():
+        validate_payload(validator, record, f"publish-record<{document_id}>")
+        artifact_ref = require_non_empty_str(
+            record.get("artifact_ref"), "record.artifact_ref"
+        )
+        normalized_ref = require_non_empty_str(
+            record.get("normalized_ref"), "record.normalized_ref"
+        )
+        classified_ref = require_non_empty_str(
+            record.get("classified_ref"), "record.classified_ref"
+        )
+        chunk_ref = require_non_empty_str(record.get("chunk_ref"), "record.chunk_ref")
+        for relative_path in (artifact_ref, normalized_ref, classified_ref, chunk_ref):
+            absolute_path = deploy_root / relative_path
+            if not absolute_path.exists():
+                raise KnowledgeContractError(
+                    f"publish record 引用了不存在的文件: {absolute_path}"
+                )
+
+    print(f"knowledge publish record 校验通过 ({len(records)} 条)")
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except KnowledgeContractError as exc:
+        print(str(exc))
+        raise SystemExit(1) from exc

scripts/verify.sh

@@ -31,6 +31,9 @@ uvx --with pyyaml --with jsonschema python scripts/evidence/validate.py
 uvx --with pyyaml --with jsonschema python scripts/evidence/collect.py --out /tmp/evidence-index.json --summary /tmp/evidence-latest.md
 uvx --with pyyaml python scripts/promoter/queue_metrics.py --queue release/queue.yaml --out /tmp/queue-health.json
 
+echo "[verify] validate knowledge publish records"
+uvx --with pyyaml --with jsonschema python -m scripts.knowledge.validate_state --deploy-root . --env dev
+
 echo "[verify] run promoter dry-run on local repo"
 python3 scripts/promoter/promote.py --dry-run --local-repo-dir .
 echo "[verify] promoter supports local simulation without Harbor via --skip-registry-check"